mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-11-07 06:19:43 -10:00
feat(dingtalk): project reserved file output safely (#9991)
* feat(dingtalk): project reserved file output safely * fix(dingtalk): close block-streaming file projection leaks (#9991) Co-authored-by: Qwen-Coder <[email protected]> * fix(dingtalk): settle file projection at turn end and segment resets (#9991) * fix(dingtalk): drop evicted-turn block sends and spurious divergence notice (#9991) Two Critical races from round 4: - /clear eviction and session death settle the block projector WITHOUT draining the turn's serialized send chain. A queued block then recreated projector state and POSTed the tail of a force-split [FILE: ...] marker verbatim. Arm block projection in onPromptStart and disarm at settle / session death: a block send finding no state while unarmed is dropped. Regression tests drive the real ChannelBase /clear machinery with a wedged turn and a hung webhook send. - The whole-turn hash divergence check mis-fired on every routine multi-tool turn: both bridges return only post-last-boundary chunks as the final text while the run-keyed projector sees every chunk, so any turn streaming text before a tool call got a spurious [File delivery unavailable]. Drop the clause and the now-dead matches()/hash state; streamed-marker fail-closed is preserved by the monotonic marker count. Also addresses the round's suggestions: bound sendReply's webhook POST with a 15s AbortSignal timeout matching the package's other fetches (the root cause behind the unbounded settle drain), document the best-effort settle tail send, extract the background-response delivery step into a ChannelBase hook so the DingTalk override no longer duplicates the base dispatch flow, cover the group proactive and silent-drop branches, assert fallback delivery bodies, log sanitized redaction breadcrumbs, and record the raw-text-scan policy on projectFileText. --------- Co-authored-by: qqqys <[email protected]> Co-authored-by: qwen-code-dev-bot <[email protected]> Co-authored-by: qwen-code-bot <[email protected]> Co-authored-by: Qwen-Coder <[email protected]>
This commit is contained in:
10 files changed
+1567
-27
No files matched your search
@@ -42,9 +42,13 @@ describe('BlockStreamer', () => {
|
||||
it('emits at paragraph boundary when buffer >= minChars', async () => {
|
||||
const s = createStreamer({ minChars: 10 });
|
||||
s.push('Hello world, this is a paragraph.\n\nSecond part');
|
||||
// Should have emitted the first paragraph
|
||||
// Should have emitted the first paragraph, keeping the line terminator
|
||||
// trim removed so consumers can see the real line boundary.
|
||||
await s.flush();
|
||||
expect(sent).toEqual(['Hello world, this is a paragraph.', 'Second part']);
|
||||
expect(sent).toEqual([
|
||||
'Hello world, this is a paragraph.\n',
|
||||
'Second part',
|
||||
]);
|
||||
expect(s.blockCount).toBe(2);
|
||||
});
|
||||
|
||||
@@ -82,7 +86,7 @@ describe('BlockStreamer', () => {
|
||||
s.push('line one\n\nline two\nline three xx');
|
||||
await s.flush();
|
||||
// Should split at \n\n (pos 10) since it's within maxChars
|
||||
expect(sent[0]).toBe('line one');
|
||||
expect(sent[0]).toBe('line one\n');
|
||||
expect(sent.length).toBe(2);
|
||||
});
|
||||
|
||||
@@ -151,8 +155,9 @@ describe('BlockStreamer', () => {
|
||||
const s = createStreamer({ minChars: 5 });
|
||||
s.push(' \n Hello world \n\n Next ');
|
||||
await s.flush();
|
||||
// The first block includes leading whitespace up to \n\n, trimmed
|
||||
expect(sent.every((t) => t === t.trim())).toBe(true);
|
||||
// Leading whitespace and trailing spaces are trimmed; a block ending at
|
||||
// a line break keeps exactly one restored newline.
|
||||
expect(sent).toEqual(['Hello world\n', 'Next']);
|
||||
});
|
||||
|
||||
it('does not emit empty blocks after trimming', async () => {
|
||||
@@ -197,7 +202,40 @@ describe('BlockStreamer', () => {
|
||||
s.push('Para one.\n\nPara two.\n\nPara three.');
|
||||
await s.flush();
|
||||
// Should emit paras 1+2 as one block (last \n\n boundary), then para 3
|
||||
expect(sent).toEqual(['Para one.\n\nPara two.', 'Para three.']);
|
||||
expect(sent).toEqual(['Para one.\n\nPara two.\n', 'Para three.']);
|
||||
});
|
||||
|
||||
it('restores one trailing newline when a block ends at a line break', async () => {
|
||||
const s = createStreamer({ minChars: 10 });
|
||||
s.push('First paragraph of the answer.\n\nSecond part');
|
||||
await s.flush();
|
||||
expect(sent).toEqual(['First paragraph of the answer.\n', 'Second part']);
|
||||
});
|
||||
|
||||
it('adds no newline when a block is force-split mid-line', async () => {
|
||||
const s = createStreamer({ minChars: 5, maxChars: 12 });
|
||||
s.push('word one word two word three');
|
||||
await s.flush();
|
||||
expect(sent[0]).not.toMatch(/\n$/);
|
||||
});
|
||||
|
||||
it('drain awaits queued sends without emitting buffered text', async () => {
|
||||
vi.useRealTimers();
|
||||
const delivered: string[] = [];
|
||||
const s = new BlockStreamer({
|
||||
minChars: 5,
|
||||
maxChars: 10,
|
||||
idleMs: 0,
|
||||
send: async (text) => {
|
||||
await new Promise((r) => setTimeout(r, 5));
|
||||
delivered.push(text);
|
||||
},
|
||||
});
|
||||
s.push('aaaa bbbb cccc dddd eeee');
|
||||
await s.drain();
|
||||
expect(delivered).toEqual(['aaaa bbbb', 'cccc dddd']);
|
||||
await s.flush();
|
||||
expect(delivered).toEqual(['aaaa bbbb', 'cccc dddd', 'eeee']);
|
||||
});
|
||||
|
||||
it('works with idleMs=0 (idle timer disabled)', async () => {
|
||||
|
||||
@@ -22,7 +22,10 @@ export interface BlockStreamerOptions {
|
||||
maxChars: number;
|
||||
/** Emit buffered text after this many ms of inactivity. Default: 1500. */
|
||||
idleMs: number;
|
||||
/** Callback to deliver a completed block. Called with trimmed text. */
|
||||
/**
|
||||
* Callback to deliver a completed block. Called with trimmed text, plus
|
||||
* one restored trailing newline when the block ended at a line break.
|
||||
*/
|
||||
send: (text: string) => Promise<void>;
|
||||
}
|
||||
|
||||
@@ -66,6 +69,11 @@ export class BlockStreamer {
|
||||
this.buffer = '';
|
||||
}
|
||||
|
||||
/** Await already-queued sends without emitting the buffered text. */
|
||||
async drain(): Promise<void> {
|
||||
await this.sending;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Internal
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -99,9 +107,13 @@ export class BlockStreamer {
|
||||
private emitBlock(text: string): void {
|
||||
const trimmed = text.trim();
|
||||
if (!trimmed) return;
|
||||
// Restore the line terminator trim removed when the block ended at one:
|
||||
// consumers that track line boundaries (redaction projectors) must be
|
||||
// able to close the line the real newline ended.
|
||||
const block = /\n[^\S\n]*$/.test(text) ? `${trimmed}\n` : trimmed;
|
||||
this.blockCount++;
|
||||
this.sending = this.sending
|
||||
.then(() => this.opts.send(trimmed))
|
||||
.then(() => this.opts.send(block))
|
||||
.catch(() => {});
|
||||
}
|
||||
|
||||
|
||||
@@ -286,6 +286,30 @@ class ResponseTrackingChannel extends TestChannel {
|
||||
}
|
||||
}
|
||||
|
||||
class SlowBlockSendChannel extends TestChannel {
|
||||
sendCompletions = 0;
|
||||
completionsAtPromptEnd: number[] = [];
|
||||
|
||||
protected override async sendResponseMessage(
|
||||
chatId: string,
|
||||
text: string,
|
||||
sessionId: string,
|
||||
): Promise<void> {
|
||||
await new Promise((resolve) => setTimeout(resolve, 15));
|
||||
this.sendCompletions++;
|
||||
await super.sendResponseMessage(chatId, text, sessionId);
|
||||
}
|
||||
|
||||
protected override onPromptEnd(
|
||||
chatId: string,
|
||||
sessionId: string,
|
||||
messageId?: string,
|
||||
): void {
|
||||
this.completionsAtPromptEnd.push(this.sendCompletions);
|
||||
super.onPromptEnd(chatId, sessionId, messageId);
|
||||
}
|
||||
}
|
||||
|
||||
class UnsafeProcessChannel extends TestChannel {
|
||||
processWithoutPreflight(envelope: Envelope): Promise<void> {
|
||||
return this.processInbound(envelope);
|
||||
@@ -13527,6 +13551,34 @@ describe('ChannelBase', () => {
|
||||
]);
|
||||
});
|
||||
|
||||
it('settles turn cleanup only after queued block sends land', async () => {
|
||||
(bridge.prompt as ReturnType<typeof vi.fn>).mockImplementation(
|
||||
(sid: string) => {
|
||||
(bridge as unknown as EventEmitter).emit(
|
||||
'textChunk',
|
||||
sid,
|
||||
'first paragraph body\n\n',
|
||||
);
|
||||
return Promise.reject(new Error('agent boom'));
|
||||
},
|
||||
);
|
||||
const ch = new SlowBlockSendChannel(
|
||||
'test-chan',
|
||||
defaultConfig({
|
||||
blockStreaming: 'on',
|
||||
blockStreamingChunk: { minChars: 5, maxChars: 100 },
|
||||
blockStreamingCoalesce: { idleMs: 0 },
|
||||
}),
|
||||
bridge,
|
||||
);
|
||||
|
||||
await expect(ch.handleInbound(envelope())).rejects.toThrow('agent boom');
|
||||
|
||||
// The failed turn's queued block send must have completed before
|
||||
// onPromptEnd settled turn-scoped adapter state.
|
||||
expect(ch.completionsAtPromptEnd).toEqual([1]);
|
||||
});
|
||||
|
||||
it('uses block streamer when blockStreaming=on', async () => {
|
||||
// The streamer sends blocks; onResponseComplete is NOT called
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
|
||||
@@ -515,7 +515,21 @@ export abstract class ChannelBase {
|
||||
await this.pushProactive(target, text);
|
||||
return;
|
||||
}
|
||||
await this.sendResponseMessage(target.chatId, text, sessionId);
|
||||
await this.deliverBackgroundReply(target.chatId, text, sessionId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Fallback delivery of a background response when proactive send is
|
||||
* unavailable. Adapters whose turn replies bypass sendResponseMessage (to
|
||||
* stay out of turn-scoped streaming state, for example) override only this
|
||||
* step instead of re-implementing the whole dispatch flow.
|
||||
*/
|
||||
protected async deliverBackgroundReply(
|
||||
chatId: string,
|
||||
text: string,
|
||||
sessionId: string,
|
||||
): Promise<void> {
|
||||
await this.sendResponseMessage(chatId, text, sessionId);
|
||||
}
|
||||
|
||||
async dispatchPermissionRequest(
|
||||
@@ -5758,7 +5772,13 @@ export abstract class ChannelBase {
|
||||
} finally {
|
||||
promptBridge.off('textChunk', onChunk);
|
||||
promptBridge.off('responseBoundary', onResponseBoundary);
|
||||
streamer?.stop();
|
||||
if (streamer) {
|
||||
streamer.stop();
|
||||
// Queued block sends belong to this turn: let them land before
|
||||
// onPromptEnd settles turn-scoped adapter state, or a send racing
|
||||
// the settle can recreate discarded state and leak unredacted text.
|
||||
await streamer.drain();
|
||||
}
|
||||
// Identity guard: a turn that wedged past /clear's bounded wait gets
|
||||
// EVICTED — /clear gives up on active.done, deletes activePrompts, and a
|
||||
// turn the user starts AFTER the clear can re-seed activePrompts (and own
|
||||
|
||||
@@ -38,7 +38,7 @@ export type {
|
||||
} from './DaemonChannelBridge.js';
|
||||
export { BlockStreamer } from './BlockStreamer.js';
|
||||
export type { BlockStreamerOptions } from './BlockStreamer.js';
|
||||
export { ChannelBase } from './ChannelBase.js';
|
||||
export { ChannelBase, CLEAR_CANCEL_TIMEOUT_MS } from './ChannelBase.js';
|
||||
export {
|
||||
CHANNEL_PROACTIVE_DELIVERY_ERROR_CODE,
|
||||
ChannelProactiveDeliveryError,
|
||||
|
||||
@@ -0,0 +1,245 @@
|
||||
import { EventEmitter } from 'node:events';
|
||||
import type { DWClientDownStream } from 'dingtalk-stream-sdk-nodejs';
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest';
|
||||
import {
|
||||
CLEAR_CANCEL_TIMEOUT_MS,
|
||||
type ChannelAgentBridge,
|
||||
type Envelope,
|
||||
} from '@qwen-code/channel-base';
|
||||
import { DingtalkChannel } from './DingtalkAdapter.js';
|
||||
|
||||
// The adapter's main test file replaces ChannelBase with a stub, so the
|
||||
// /clear eviction and session-died races can only be witnessed here, against
|
||||
// the REAL ChannelBase machinery.
|
||||
|
||||
const dingtalkSdkMock = vi.hoisted(() => ({
|
||||
instances: [] as unknown[],
|
||||
}));
|
||||
|
||||
vi.mock('dingtalk-stream-sdk-nodejs', () => ({
|
||||
DWClient: class {
|
||||
debug = true;
|
||||
connected = true;
|
||||
registered = true;
|
||||
config = { autoReconnect: true };
|
||||
socket = new (class {
|
||||
readyState = 1;
|
||||
ping = vi.fn();
|
||||
private listeners = new Map<string, Set<(...args: unknown[]) => void>>();
|
||||
|
||||
on(event: string, listener: (...args: unknown[]) => void): void {
|
||||
const listeners = this.listeners.get(event) ?? new Set();
|
||||
listeners.add(listener);
|
||||
this.listeners.set(event, listeners);
|
||||
}
|
||||
|
||||
off(event: string, listener: (...args: unknown[]) => void): void {
|
||||
this.listeners.get(event)?.delete(listener);
|
||||
}
|
||||
|
||||
emit(event: string, ...args: unknown[]): void {
|
||||
for (const listener of this.listeners.get(event) ?? []) {
|
||||
listener(...args);
|
||||
}
|
||||
}
|
||||
})();
|
||||
callback?: (msg: DWClientDownStream) => void;
|
||||
callbacks = new Map<string, (msg: DWClientDownStream) => void>();
|
||||
disconnect = vi.fn();
|
||||
getConfig = vi.fn(() => ({ access_token: 'token' }));
|
||||
registerCallbackListener = vi.fn(
|
||||
(topic: string, callback: (msg: DWClientDownStream) => void) => {
|
||||
this.callbacks.set(topic, callback);
|
||||
if (topic === 'robot') this.callback = callback;
|
||||
},
|
||||
);
|
||||
send = vi.fn();
|
||||
connect = vi.fn(() => Promise.resolve());
|
||||
|
||||
onSystem = vi.fn();
|
||||
onEvent = vi.fn();
|
||||
onCallback = vi.fn();
|
||||
onDownStream = vi.fn();
|
||||
|
||||
constructor(readonly options: Record<string, unknown>) {
|
||||
dingtalkSdkMock.instances.push(this);
|
||||
}
|
||||
},
|
||||
TOPIC_ROBOT: 'robot',
|
||||
TOPIC_CARD: 'card',
|
||||
EventAck: { SUCCESS: 'success' },
|
||||
}));
|
||||
|
||||
function createBridge(promptImpl: (sessionId: string) => Promise<string>) {
|
||||
const emitter = new EventEmitter();
|
||||
const bridge = Object.assign(emitter, {
|
||||
newSession: vi.fn().mockReturnValue('session-1'),
|
||||
loadSession: vi.fn(),
|
||||
prompt: vi.fn().mockImplementation(promptImpl),
|
||||
cancelSession: vi.fn().mockResolvedValue(undefined),
|
||||
discardSession: vi.fn().mockResolvedValue(undefined),
|
||||
stop: vi.fn(),
|
||||
start: vi.fn(),
|
||||
isConnected: true,
|
||||
availableCommands: [],
|
||||
setBridge: vi.fn(),
|
||||
respondToPermission: vi.fn().mockResolvedValue(true),
|
||||
registerChannelLoopToolHandler: vi.fn(),
|
||||
});
|
||||
return bridge as unknown as ChannelAgentBridge & {
|
||||
prompt: ReturnType<typeof vi.fn>;
|
||||
};
|
||||
}
|
||||
|
||||
function envelope(overrides: Partial<Envelope> = {}): Envelope {
|
||||
return {
|
||||
channelName: 'test-dingtalk',
|
||||
senderId: 'user1',
|
||||
senderName: 'User 1',
|
||||
chatId: 'cid123',
|
||||
text: 'hello',
|
||||
isGroup: false,
|
||||
isMentioned: false,
|
||||
isReplyToBot: false,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
function createChannel(bridge: ChannelAgentBridge): DingtalkChannel {
|
||||
return new DingtalkChannel(
|
||||
'test-dingtalk',
|
||||
{
|
||||
type: 'dingtalk',
|
||||
token: 'tok',
|
||||
clientId: 'client-id',
|
||||
clientSecret: 'client-secret',
|
||||
senderPolicy: 'open',
|
||||
allowedUsers: [],
|
||||
sessionScope: 'user',
|
||||
cwd: '/tmp',
|
||||
groupPolicy: 'disabled',
|
||||
dmPolicy: 'open',
|
||||
groups: {},
|
||||
blockStreaming: 'on',
|
||||
blockStreamingChunk: { minChars: 5, maxChars: 10 },
|
||||
blockStreamingCoalesce: { idleMs: 60_000 },
|
||||
} as never,
|
||||
bridge,
|
||||
);
|
||||
}
|
||||
|
||||
function seedWebhook(channel: DingtalkChannel, chatId: string): void {
|
||||
(channel as unknown as { webhooks: Map<string, string> }).webhooks.set(
|
||||
chatId,
|
||||
'https://oapi.dingtalk.com/robot/send?access_token=token',
|
||||
);
|
||||
}
|
||||
|
||||
// maxChars=10 splits this into 'aaaa\n', '[FILE:', '/workspace',
|
||||
// '/secret.tx', 't] more'. Hanging the SECOND send leaves the opener
|
||||
// projected into the turn's projector (reserved line held) with the path
|
||||
// bytes still queued — exactly the blocks that race the settle.
|
||||
const MARKER_TEXT = 'aaaa\n\n[FILE: /workspace/secret.txt] more more';
|
||||
|
||||
describe('DingtalkChannel block projection under eviction', () => {
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it('/clear on a wedged turn drops the late blocks instead of leaking the marker tail', async () => {
|
||||
const bridge = createBridge((sessionId) => {
|
||||
(bridge as unknown as EventEmitter).emit(
|
||||
'textChunk',
|
||||
sessionId,
|
||||
MARKER_TEXT,
|
||||
);
|
||||
return new Promise<string>(() => {});
|
||||
});
|
||||
const channel = createChannel(bridge);
|
||||
seedWebhook(channel, 'cid123');
|
||||
|
||||
const bodies: string[] = [];
|
||||
let fetchCalls = 0;
|
||||
const pendingFetch: Array<(response: Response) => void> = [];
|
||||
vi.spyOn(globalThis, 'fetch').mockImplementation((_url, init) => {
|
||||
bodies.push(String((init as RequestInit | undefined)?.body ?? ''));
|
||||
fetchCalls += 1;
|
||||
// Hang exactly the opener block's send: it is projected (reserved line
|
||||
// held) before settle, while the path bytes wait behind it in the
|
||||
// serialized send chain.
|
||||
if (fetchCalls === 2) {
|
||||
return new Promise<Response>((resolve) => pendingFetch.push(resolve));
|
||||
}
|
||||
return Promise.resolve(new Response('{}'));
|
||||
});
|
||||
|
||||
void channel.handleInbound(envelope({ messageId: 'm1' }));
|
||||
// Block 1 landed; the opener block's send hangs in fetch with the path
|
||||
// blocks queued behind it.
|
||||
await vi.waitFor(() => expect(bodies).toHaveLength(2));
|
||||
|
||||
vi.useFakeTimers();
|
||||
const clearPromise = channel.handleInbound(
|
||||
envelope({ messageId: 'm2', text: '/clear' }),
|
||||
);
|
||||
// Drive the bounded wind-down wait to eviction: settle runs while the
|
||||
// path blocks are still queued in the turn's send chain.
|
||||
await vi.advanceTimersByTimeAsync(CLEAR_CANCEL_TIMEOUT_MS);
|
||||
for (const resolve of pendingFetch.splice(0)) {
|
||||
resolve(new Response('{}'));
|
||||
}
|
||||
vi.useRealTimers();
|
||||
await clearPromise;
|
||||
// Let the detached send chain drain its now-unblocked sends.
|
||||
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
|
||||
const all = bodies.join('\n');
|
||||
expect(all).toContain('File delivery unavailable');
|
||||
expect(all).not.toContain('secret');
|
||||
expect(all).not.toContain('[FILE:');
|
||||
expect(
|
||||
(channel as unknown as { blockFileProjectors: Map<string, unknown> })
|
||||
.blockFileProjectors.size,
|
||||
).toBe(0);
|
||||
});
|
||||
|
||||
it('drops the late blocks when the session dies mid-send', async () => {
|
||||
const bridge = createBridge((sessionId) => {
|
||||
(bridge as unknown as EventEmitter).emit(
|
||||
'textChunk',
|
||||
sessionId,
|
||||
MARKER_TEXT,
|
||||
);
|
||||
return new Promise<string>(() => {});
|
||||
});
|
||||
const channel = createChannel(bridge);
|
||||
seedWebhook(channel, 'cid123');
|
||||
|
||||
const bodies: string[] = [];
|
||||
let fetchCalls = 0;
|
||||
const pendingFetch: Array<(response: Response) => void> = [];
|
||||
vi.spyOn(globalThis, 'fetch').mockImplementation((_url, init) => {
|
||||
bodies.push(String((init as RequestInit | undefined)?.body ?? ''));
|
||||
fetchCalls += 1;
|
||||
if (fetchCalls === 2) {
|
||||
return new Promise<Response>((resolve) => pendingFetch.push(resolve));
|
||||
}
|
||||
return Promise.resolve(new Response('{}'));
|
||||
});
|
||||
|
||||
void channel.handleInbound(envelope({ messageId: 'm1' }));
|
||||
await vi.waitFor(() => expect(bodies).toHaveLength(2));
|
||||
|
||||
channel.onSessionDied('session-1');
|
||||
for (const resolve of pendingFetch.splice(0)) {
|
||||
resolve(new Response('{}'));
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
|
||||
const all = bodies.join('\n');
|
||||
expect(all).toContain('File delivery unavailable');
|
||||
expect(all).not.toContain('secret');
|
||||
expect(all).not.toContain('[FILE:');
|
||||
});
|
||||
});
|
||||
@@ -11,6 +11,7 @@ import {
|
||||
import { tmpdir } from 'node:os';
|
||||
import { dirname, join } from 'node:path';
|
||||
import type { DWClientDownStream } from 'dingtalk-stream-sdk-nodejs';
|
||||
import { BlockStreamer } from '@qwen-code/channel-base';
|
||||
import type {
|
||||
ChannelOutputSegmentContext,
|
||||
ChannelOutputSegmentEndReason,
|
||||
@@ -144,6 +145,36 @@ vi.mock('@qwen-code/channel-base', async () => {
|
||||
_messageIds: string[],
|
||||
): void {}
|
||||
protected requestPromptRunCancellation = vi.fn().mockResolvedValue(false);
|
||||
// Real base dispatch flow, delegated like logDebugPayload: the adapter
|
||||
// override under test replaces only the final delivery step.
|
||||
async dispatchBackgroundResponse(
|
||||
sessionId: string,
|
||||
text: string,
|
||||
): Promise<void> {
|
||||
await (
|
||||
real.ChannelBase.prototype as unknown as {
|
||||
dispatchBackgroundResponse(
|
||||
sessionId: string,
|
||||
text: string,
|
||||
): Promise<void>;
|
||||
}
|
||||
).dispatchBackgroundResponse.call(this, sessionId, text);
|
||||
}
|
||||
protected async deliverBackgroundReply(
|
||||
chatId: string,
|
||||
text: string,
|
||||
sessionId: string,
|
||||
): Promise<void> {
|
||||
await (
|
||||
real.ChannelBase.prototype as unknown as {
|
||||
deliverBackgroundReply(
|
||||
chatId: string,
|
||||
text: string,
|
||||
sessionId: string,
|
||||
): Promise<void>;
|
||||
}
|
||||
).deliverBackgroundReply.call(this, chatId, text, sessionId);
|
||||
}
|
||||
protected supportsProactiveTarget(target: SessionTarget): boolean {
|
||||
return target.threadId === undefined;
|
||||
}
|
||||
@@ -171,6 +202,9 @@ vi.mock('@qwen-code/channel-base', async () => {
|
||||
// ship green.
|
||||
truncateUtf16Units: real.truncateUtf16Units,
|
||||
isTerminalTaskLifecycleType: real.isTerminalTaskLifecycleType,
|
||||
// Real: the block-boundary regression drives blocks through the actual
|
||||
// streamer's trim contract, not a hand-built block shape.
|
||||
BlockStreamer: real.BlockStreamer,
|
||||
};
|
||||
});
|
||||
|
||||
@@ -899,6 +933,20 @@ function seedMentionTarget(
|
||||
).mentionTargets.set(messageId, staffId);
|
||||
}
|
||||
|
||||
function seedSessionTarget(
|
||||
channel: DingtalkChannelInstance,
|
||||
sessionId: string,
|
||||
target: SessionTarget,
|
||||
): void {
|
||||
(
|
||||
channel as unknown as {
|
||||
router: { getTarget(sessionId: string): SessionTarget | undefined };
|
||||
}
|
||||
).router = {
|
||||
getTarget: (id: string) => (id === sessionId ? target : undefined),
|
||||
};
|
||||
}
|
||||
|
||||
function deferredPromise<T>() {
|
||||
let resolve!: (value: T | PromiseLike<T>) => void;
|
||||
let reject!: (reason?: unknown) => void;
|
||||
@@ -1824,6 +1872,52 @@ describe('DingtalkChannel status cards', () => {
|
||||
segmentId: 'segment-2',
|
||||
});
|
||||
expect(fetchSpy).toHaveBeenCalledOnce();
|
||||
const fallbackBody = JSON.parse(
|
||||
String((fetchSpy.mock.calls[0]![1] as RequestInit).body),
|
||||
) as { markdown: { text: string } };
|
||||
expect(fallbackBody.markdown.text).toBe('second');
|
||||
});
|
||||
|
||||
it('falls back to the reply delivery with the projected text when no presenter exists', async () => {
|
||||
const channel = createChannel();
|
||||
seedWebhook(channel, 'cid-1');
|
||||
const fetchSpy = vi
|
||||
.spyOn(globalThis, 'fetch')
|
||||
.mockResolvedValue(new Response('{}'));
|
||||
const context = {
|
||||
channelName: 'dingtalk',
|
||||
sessionId: 'session-1',
|
||||
runId: 'run-1',
|
||||
segmentId: 'segment-1',
|
||||
owner: { kind: 'channel_user', id: 'owner-1' },
|
||||
target: {
|
||||
channelName: 'dingtalk',
|
||||
chatId: 'cid-1',
|
||||
senderId: 'owner-1',
|
||||
isGroup: true,
|
||||
},
|
||||
} as ChannelOutputSegmentContext;
|
||||
|
||||
getChunkHook(channel)(
|
||||
'cid-1',
|
||||
'[FILE: /workspace/a.txt]\npartial answer',
|
||||
'session-1',
|
||||
context,
|
||||
);
|
||||
await getCompleteHook(channel)(
|
||||
'cid-1',
|
||||
'final answer',
|
||||
'session-1',
|
||||
context,
|
||||
);
|
||||
|
||||
expect(fetchSpy).toHaveBeenCalledOnce();
|
||||
const body = JSON.parse(
|
||||
String((fetchSpy.mock.calls[0]![1] as RequestInit).body),
|
||||
) as { markdown: { text: string } };
|
||||
expect(body.markdown.text).toContain('final answer');
|
||||
expect(body.markdown.text).toContain('[File delivery unavailable]');
|
||||
expect(body.markdown.text).not.toContain('/workspace/a.txt');
|
||||
});
|
||||
|
||||
it('uploads a final status card image before closing output', async () => {
|
||||
@@ -5854,6 +5948,703 @@ describe('DingtalkChannel outbound image delivery', () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe('DingtalkChannel outbound file projection', () => {
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
function segment(segmentId = 'segment-1'): ChannelOutputSegmentContext {
|
||||
return {
|
||||
channelName: 'dingtalk',
|
||||
sessionId: 'session-1',
|
||||
runId: 'run-1',
|
||||
segmentId,
|
||||
owner: { kind: 'channel_user', id: 'owner-1' },
|
||||
target: {
|
||||
channelName: 'dingtalk',
|
||||
chatId: 'cid123',
|
||||
senderId: 'owner-1',
|
||||
isGroup: true,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
it('redacts reserved file output from plain replies', async () => {
|
||||
const channel = createChannel();
|
||||
seedWebhook(channel, 'cid123');
|
||||
const fetchSpy = vi
|
||||
.spyOn(globalThis, 'fetch')
|
||||
.mockResolvedValue(new Response('{}'));
|
||||
|
||||
await channel.sendMessage(
|
||||
'cid123',
|
||||
'before\n[FILE: /workspace/report.txt]\nafter',
|
||||
);
|
||||
|
||||
const body = JSON.parse(
|
||||
String((fetchSpy.mock.calls[0]![1] as RequestInit).body),
|
||||
) as { markdown: { text: string } };
|
||||
expect(body.markdown.text).toContain('before\n\nafter');
|
||||
expect(body.markdown.text).toContain('[File delivery unavailable]');
|
||||
expect(body.markdown.text).not.toContain('[FILE:');
|
||||
expect(body.markdown.text).not.toContain('/workspace/report.txt');
|
||||
});
|
||||
|
||||
it.each([
|
||||
['reserved opening', '[FILE:', ''],
|
||||
['split reserved opening', '[FI', 'LE: '],
|
||||
])(
|
||||
'keeps paths hidden when block streaming splits the %s',
|
||||
async (_name, first, second) => {
|
||||
const channel = createChannel({ blockStreaming: 'on' });
|
||||
seedWebhook(channel, 'cid123');
|
||||
getPromptHook(channel, 'onPromptStart')('cid123', 'session-1');
|
||||
const fetchSpy = vi
|
||||
.spyOn(globalThis, 'fetch')
|
||||
.mockResolvedValue(new Response('{}'));
|
||||
const send = (
|
||||
channel as unknown as {
|
||||
sendResponseMessage(
|
||||
chatId: string,
|
||||
text: string,
|
||||
sessionId: string,
|
||||
): Promise<void>;
|
||||
}
|
||||
).sendResponseMessage.bind(channel);
|
||||
|
||||
await send('cid123', first, 'session-1');
|
||||
await send(
|
||||
'cid123',
|
||||
`${second}/workspace/private-report.txt]`,
|
||||
'session-1',
|
||||
);
|
||||
|
||||
expect(JSON.stringify(fetchSpy.mock.calls)).not.toContain(
|
||||
'/workspace/private-report.txt',
|
||||
);
|
||||
expect(JSON.stringify(fetchSpy.mock.calls)).toContain(
|
||||
'File delivery unavailable',
|
||||
);
|
||||
await getOutputSegmentEndHook(channel)(
|
||||
'cid123',
|
||||
'session-1',
|
||||
segment(),
|
||||
'completed',
|
||||
);
|
||||
expect(
|
||||
(channel as unknown as { blockFileProjectors: Map<string, unknown> })
|
||||
.blockFileProjectors.size,
|
||||
).toBe(1);
|
||||
getPromptHook(channel, 'onPromptEnd')('cid123', 'session-1');
|
||||
expect(
|
||||
(channel as unknown as { blockFileProjectors: Map<string, unknown> })
|
||||
.blockFileProjectors.size,
|
||||
).toBe(0);
|
||||
},
|
||||
);
|
||||
|
||||
it('keeps the block projector across a segment reset so split markers stay redacted', async () => {
|
||||
const channel = createChannel({ blockStreaming: 'on' });
|
||||
seedWebhook(channel, 'cid123');
|
||||
getPromptHook(channel, 'onPromptStart')('cid123', 'session-1');
|
||||
const fetchSpy = vi
|
||||
.spyOn(globalThis, 'fetch')
|
||||
.mockResolvedValue(new Response('{}'));
|
||||
const send = getResponseHook(channel);
|
||||
|
||||
await send('cid123', 'Report\n[FILE: /workspace/secret-', 'session-1');
|
||||
await getOutputSegmentEndHook(channel)(
|
||||
'cid123',
|
||||
'session-1',
|
||||
segment(),
|
||||
'response_boundary',
|
||||
);
|
||||
await send('cid123', 'report.txt]\nDone', 'session-1');
|
||||
|
||||
const texts = fetchSpy.mock.calls.map(
|
||||
([, init]) =>
|
||||
(
|
||||
JSON.parse(String((init as RequestInit).body)) as {
|
||||
markdown: { text: string };
|
||||
}
|
||||
).markdown.text,
|
||||
);
|
||||
expect(texts.join('\n')).not.toContain('report.txt');
|
||||
expect(texts.join('\n')).toContain('File delivery unavailable');
|
||||
});
|
||||
|
||||
it('keeps a reserved line pending across blocks that end on an early "]"', async () => {
|
||||
const channel = createChannel({ blockStreaming: 'on' });
|
||||
seedWebhook(channel, 'cid123');
|
||||
getPromptHook(channel, 'onPromptStart')('cid123', 'session-1');
|
||||
const fetchSpy = vi
|
||||
.spyOn(globalThis, 'fetch')
|
||||
.mockResolvedValue(new Response('{}'));
|
||||
const send = getResponseHook(channel);
|
||||
|
||||
await send('cid123', 'before\n[FILE: /workspace/report [v2]', 'session-1');
|
||||
await send('cid123', '.txt]\nafter', 'session-1');
|
||||
|
||||
const bodies = JSON.stringify(fetchSpy.mock.calls);
|
||||
expect(bodies).not.toContain('.txt]');
|
||||
expect(bodies).not.toContain('[FILE:');
|
||||
expect(bodies).toContain('File delivery unavailable');
|
||||
});
|
||||
|
||||
it('reports the unavailable notice once across later blocks', async () => {
|
||||
const channel = createChannel({ blockStreaming: 'on' });
|
||||
seedWebhook(channel, 'cid123');
|
||||
getPromptHook(channel, 'onPromptStart')('cid123', 'session-1');
|
||||
const fetchSpy = vi
|
||||
.spyOn(globalThis, 'fetch')
|
||||
.mockResolvedValue(new Response('{}'));
|
||||
const send = getResponseHook(channel);
|
||||
|
||||
await send('cid123', '[FILE: /workspace/a.txt]\n', 'session-1');
|
||||
await send('cid123', 'Answer part one\n', 'session-1');
|
||||
await send('cid123', 'Answer part two\n', 'session-1');
|
||||
|
||||
const texts = fetchSpy.mock.calls.map(
|
||||
([, init]) =>
|
||||
(
|
||||
JSON.parse(String((init as RequestInit).body)) as {
|
||||
markdown: { text: string };
|
||||
}
|
||||
).markdown.text,
|
||||
);
|
||||
expect(texts.join('\n').match(/File delivery unavailable/g)).toHaveLength(
|
||||
1,
|
||||
);
|
||||
});
|
||||
|
||||
it('keeps the group mention for the answer after a notice-only block', async () => {
|
||||
const channel = createChannel({ atSender: true, blockStreaming: 'on' });
|
||||
seedWebhook(channel, 'cid123');
|
||||
seedMentionTarget(channel, 'm1', 'staff-1');
|
||||
const fetchSpy = vi
|
||||
.spyOn(globalThis, 'fetch')
|
||||
.mockResolvedValue(new Response('{}'));
|
||||
getPromptHook(channel, 'onPromptStart')('cid123', 'session-1', 'm1');
|
||||
const send = getResponseHook(channel);
|
||||
|
||||
await send('cid123', '[FILE: /workspace/a.txt', 'session-1');
|
||||
await send('cid123', ']\nThe answer', 'session-1');
|
||||
|
||||
const bodies = fetchSpy.mock.calls.map(([, init]) =>
|
||||
JSON.parse(String((init as RequestInit).body)),
|
||||
) as Array<{ markdown: { text: string }; at?: { atUserIds: string[] } }>;
|
||||
expect(bodies).toHaveLength(2);
|
||||
expect(bodies[0]!.markdown.text).toBe('[File delivery unavailable]');
|
||||
expect(bodies[0]).not.toHaveProperty('at');
|
||||
expect(bodies[1]!.markdown.text).toContain('@staff-1');
|
||||
expect(bodies[1]!.markdown.text).toContain('The answer');
|
||||
expect(bodies[1]!.at).toEqual({ atUserIds: ['staff-1'] });
|
||||
});
|
||||
|
||||
it('delivers DM background responses without interleaving the block projector', async () => {
|
||||
const channel = createChannel({ blockStreaming: 'on' });
|
||||
seedWebhook(channel, 'cid123');
|
||||
getPromptHook(channel, 'onPromptStart')('cid123', 'session-1');
|
||||
seedSessionTarget(channel, 'session-1', {
|
||||
channelName: 'test-dingtalk',
|
||||
senderId: 'user-1',
|
||||
chatId: 'cid123',
|
||||
isGroup: false,
|
||||
});
|
||||
const fetchSpy = vi
|
||||
.spyOn(globalThis, 'fetch')
|
||||
.mockResolvedValue(new Response('{}'));
|
||||
const send = getResponseHook(channel);
|
||||
|
||||
await send(
|
||||
'cid123',
|
||||
'Partial answer [FILE: /workspace/report.txt',
|
||||
'session-1',
|
||||
);
|
||||
await channel.dispatchBackgroundResponse(
|
||||
'session-1',
|
||||
'Background notification',
|
||||
);
|
||||
|
||||
const bodies = fetchSpy.mock.calls.map(([, init]) =>
|
||||
JSON.parse(String((init as RequestInit).body)),
|
||||
) as Array<{ markdown: { text: string } }>;
|
||||
expect(bodies).toHaveLength(2);
|
||||
expect(bodies[1]!.markdown.text).toBe('Background notification');
|
||||
expect(JSON.stringify(bodies)).not.toContain('[FILE:');
|
||||
expect(JSON.stringify(bodies)).not.toContain('/workspace/report.txt');
|
||||
});
|
||||
|
||||
it('delivers group background responses proactively, past the block projector', async () => {
|
||||
const channel = createChannel({ blockStreaming: 'on' });
|
||||
seedWebhook(channel, 'cid123');
|
||||
seedSessionTarget(channel, 'session-1', {
|
||||
channelName: 'test-dingtalk',
|
||||
senderId: 'user-1',
|
||||
chatId: 'cidGroup==',
|
||||
isGroup: true,
|
||||
});
|
||||
const fetchSpy = vi
|
||||
.spyOn(globalThis, 'fetch')
|
||||
.mockResolvedValue(new Response('{}'));
|
||||
const pushProactive = vi
|
||||
.spyOn(
|
||||
channel as unknown as {
|
||||
pushProactive(target: SessionTarget, text: string): Promise<void>;
|
||||
},
|
||||
'pushProactive',
|
||||
)
|
||||
.mockResolvedValue(undefined);
|
||||
|
||||
getPromptHook(channel, 'onPromptStart')('cid123', 'session-1');
|
||||
await getResponseHook(channel)(
|
||||
'cid123',
|
||||
'Partial [FILE: /workspace/report.txt',
|
||||
'session-1',
|
||||
);
|
||||
|
||||
await channel.dispatchBackgroundResponse(
|
||||
'session-1',
|
||||
'Background notification',
|
||||
);
|
||||
|
||||
expect(pushProactive).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ chatId: 'cidGroup==' }),
|
||||
'Background notification',
|
||||
);
|
||||
// The notification bypassed sendReply entirely; only the turn's own
|
||||
// block reached the webhook, and the held marker stayed in the projector.
|
||||
expect(fetchSpy).toHaveBeenCalledOnce();
|
||||
expect(
|
||||
(channel as unknown as { blockFileProjectors: Map<string, unknown> })
|
||||
.blockFileProjectors.size,
|
||||
).toBe(1);
|
||||
});
|
||||
|
||||
it.each([
|
||||
['an unknown session', 'other-session', 'test-dingtalk'],
|
||||
['a foreign-channel target', 'session-1', 'other-channel'],
|
||||
])(
|
||||
'silently drops a background response for %s',
|
||||
async (_name, seededSession, channelName) => {
|
||||
const channel = createChannel({ blockStreaming: 'on' });
|
||||
seedSessionTarget(channel, seededSession, {
|
||||
channelName,
|
||||
senderId: 'user-1',
|
||||
chatId: 'cidGroup==',
|
||||
isGroup: true,
|
||||
});
|
||||
const fetchSpy = vi
|
||||
.spyOn(globalThis, 'fetch')
|
||||
.mockResolvedValue(new Response('{}'));
|
||||
const pushProactive = vi
|
||||
.spyOn(
|
||||
channel as unknown as {
|
||||
pushProactive(target: SessionTarget, text: string): Promise<void>;
|
||||
},
|
||||
'pushProactive',
|
||||
)
|
||||
.mockResolvedValue(undefined);
|
||||
|
||||
await channel.dispatchBackgroundResponse(
|
||||
'session-1',
|
||||
'Background notification',
|
||||
);
|
||||
|
||||
expect(pushProactive).not.toHaveBeenCalled();
|
||||
expect(fetchSpy).not.toHaveBeenCalled();
|
||||
},
|
||||
);
|
||||
|
||||
it('silently drops an empty background response', async () => {
|
||||
const channel = createChannel({ blockStreaming: 'on' });
|
||||
seedSessionTarget(channel, 'session-1', {
|
||||
channelName: 'test-dingtalk',
|
||||
senderId: 'user-1',
|
||||
chatId: 'cidGroup==',
|
||||
isGroup: true,
|
||||
});
|
||||
const fetchSpy = vi
|
||||
.spyOn(globalThis, 'fetch')
|
||||
.mockResolvedValue(new Response('{}'));
|
||||
const pushProactive = vi
|
||||
.spyOn(
|
||||
channel as unknown as {
|
||||
pushProactive(target: SessionTarget, text: string): Promise<void>;
|
||||
},
|
||||
'pushProactive',
|
||||
)
|
||||
.mockResolvedValue(undefined);
|
||||
|
||||
await channel.dispatchBackgroundResponse('session-1', ' ');
|
||||
|
||||
expect(pushProactive).not.toHaveBeenCalled();
|
||||
expect(fetchSpy).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('feeds status presentation only projected chunks and final text', async () => {
|
||||
const channel = createChannel();
|
||||
const projected: string[] = [];
|
||||
const closeOutput = vi.fn().mockResolvedValue(true);
|
||||
(
|
||||
channel as unknown as {
|
||||
interactionPresenter: {
|
||||
appendOutput: (_segment: unknown, chunk: string) => void;
|
||||
closeOutput: typeof closeOutput;
|
||||
};
|
||||
}
|
||||
).interactionPresenter = {
|
||||
appendOutput: (_segment, chunk) => projected.push(chunk),
|
||||
closeOutput,
|
||||
};
|
||||
const context = segment();
|
||||
const chunks = ['before\n[FI', 'LE: /workspace/report.txt]', '\nafter'];
|
||||
for (const chunk of chunks) {
|
||||
getChunkHook(channel)('cid123', chunk, 'session-1', context);
|
||||
}
|
||||
await getCompleteHook(channel)(
|
||||
'cid123',
|
||||
chunks.join(''),
|
||||
'session-1',
|
||||
context,
|
||||
);
|
||||
|
||||
expect(projected.join('')).toBe('before\n\nafter');
|
||||
expect(closeOutput.mock.calls[0]?.[1]).toBe(
|
||||
'before\n\nafter\n[File delivery unavailable]',
|
||||
);
|
||||
expect(JSON.stringify(closeOutput.mock.calls)).not.toContain(
|
||||
'/workspace/report.txt',
|
||||
);
|
||||
});
|
||||
|
||||
it('adds no notice when a marker-free final text differs from the streamed prefix', async () => {
|
||||
// A routine multi-tool turn: visible text streams, a response boundary
|
||||
// resets the bridge's chunk accumulation, and the final text carries only
|
||||
// the post-boundary bytes. The streamed projector saw MORE than the final
|
||||
// text by construction — that alone must not fail closed.
|
||||
const channel = createChannel();
|
||||
const closeOutput = vi.fn().mockResolvedValue(true);
|
||||
(
|
||||
channel as unknown as {
|
||||
interactionPresenter: {
|
||||
appendOutput: () => void;
|
||||
closeOutput: typeof closeOutput;
|
||||
};
|
||||
}
|
||||
).interactionPresenter = { appendOutput: () => {}, closeOutput };
|
||||
const first = segment('segment-1');
|
||||
getChunkHook(channel)(
|
||||
'cid123',
|
||||
'Sure, let me check that.',
|
||||
'session-1',
|
||||
first,
|
||||
);
|
||||
await getOutputSegmentEndHook(channel)(
|
||||
'cid123',
|
||||
'session-1',
|
||||
first,
|
||||
'response_boundary',
|
||||
);
|
||||
const next = { ...first, segmentId: 'segment-2' };
|
||||
getChunkHook(channel)('cid123', 'The answer is 42.', 'session-1', next);
|
||||
await getCompleteHook(channel)(
|
||||
'cid123',
|
||||
'The answer is 42.',
|
||||
'session-1',
|
||||
next,
|
||||
);
|
||||
// Call 0 is the response_boundary close; the final text lands in call 1.
|
||||
expect(closeOutput.mock.calls[1]?.[1]).toBe('The answer is 42.');
|
||||
});
|
||||
|
||||
it('keeps the notice when a streamed marker is absent from the final text', async () => {
|
||||
const channel = createChannel();
|
||||
const closeOutput = vi.fn().mockResolvedValue(true);
|
||||
(
|
||||
channel as unknown as {
|
||||
interactionPresenter: {
|
||||
appendOutput: () => void;
|
||||
closeOutput: typeof closeOutput;
|
||||
};
|
||||
}
|
||||
).interactionPresenter = { appendOutput: () => {}, closeOutput };
|
||||
const context = segment();
|
||||
getChunkHook(channel)(
|
||||
'cid123',
|
||||
'[FILE: /workspace/a.txt]\nstreamed text',
|
||||
'session-1',
|
||||
context,
|
||||
);
|
||||
await getCompleteHook(channel)(
|
||||
'cid123',
|
||||
'different final text',
|
||||
'session-1',
|
||||
context,
|
||||
);
|
||||
expect(closeOutput.mock.calls[0]?.[1]).toBe(
|
||||
'different final text\n[File delivery unavailable]',
|
||||
);
|
||||
expect(JSON.stringify(closeOutput.mock.calls)).not.toContain(
|
||||
'/workspace/a.txt',
|
||||
);
|
||||
});
|
||||
|
||||
it('discards segment projectors on terminal segment ends', async () => {
|
||||
const channel = createChannel();
|
||||
for (const [index, reason] of (
|
||||
['cancelled', 'failed'] as const
|
||||
).entries()) {
|
||||
const ended = segment(`segment-${index + 1}`);
|
||||
getChunkHook(channel)(
|
||||
'cid123',
|
||||
'[FILE: /workspace/a.txt]',
|
||||
'session-1',
|
||||
ended,
|
||||
);
|
||||
await getOutputSegmentEndHook(channel)(
|
||||
'cid123',
|
||||
'session-1',
|
||||
ended,
|
||||
reason,
|
||||
);
|
||||
}
|
||||
expect(
|
||||
(channel as unknown as { fileProjectors: Map<string, unknown> })
|
||||
.fileProjectors.size,
|
||||
).toBe(0);
|
||||
});
|
||||
|
||||
it('flushes the block projector held tail when the turn ends', async () => {
|
||||
const channel = createChannel({ blockStreaming: 'on' });
|
||||
seedWebhook(channel, 'cid123');
|
||||
getPromptHook(channel, 'onPromptStart')('cid123', 'session-1');
|
||||
const fetchSpy = vi
|
||||
.spyOn(globalThis, 'fetch')
|
||||
.mockResolvedValue(new Response('{}'));
|
||||
|
||||
await getResponseHook(channel)('cid123', 'Answer ends here [', 'session-1');
|
||||
getPromptHook(channel, 'onPromptEnd')('cid123', 'session-1');
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(fetchSpy.mock.calls).toHaveLength(2);
|
||||
});
|
||||
const lastBody = JSON.parse(
|
||||
String((fetchSpy.mock.calls[1]![1] as RequestInit).body),
|
||||
) as { markdown: { text: string } };
|
||||
expect(lastBody.markdown.text).toBe('[');
|
||||
expect(lastBody.markdown.text).not.toContain('File delivery unavailable');
|
||||
expect(
|
||||
(channel as unknown as { blockFileProjectors: Map<string, unknown> })
|
||||
.blockFileProjectors.size,
|
||||
).toBe(0);
|
||||
});
|
||||
|
||||
it('redacts an unfinished marker at turn end instead of leaking a fragment', async () => {
|
||||
const channel = createChannel({ blockStreaming: 'on' });
|
||||
seedWebhook(channel, 'cid123');
|
||||
getPromptHook(channel, 'onPromptStart')('cid123', 'session-1');
|
||||
const fetchSpy = vi
|
||||
.spyOn(globalThis, 'fetch')
|
||||
.mockResolvedValue(new Response('{}'));
|
||||
|
||||
await getResponseHook(channel)(
|
||||
'cid123',
|
||||
'Report\n[FILE: /workspace/sec',
|
||||
'session-1',
|
||||
);
|
||||
getPromptHook(channel, 'onPromptEnd')('cid123', 'session-1');
|
||||
|
||||
// Settle must not emit the held reserved line: the marker never
|
||||
// completed, so nothing may follow it as a standalone message.
|
||||
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
expect(fetchSpy.mock.calls).toHaveLength(1);
|
||||
expect(JSON.stringify(fetchSpy.mock.calls)).not.toContain('/workspace/sec');
|
||||
expect(
|
||||
(channel as unknown as { blockFileProjectors: Map<string, unknown> })
|
||||
.blockFileProjectors.size,
|
||||
).toBe(0);
|
||||
});
|
||||
|
||||
it('keeps the status projector across a mid-turn segment reset', async () => {
|
||||
const channel = createChannel();
|
||||
const projected: string[] = [];
|
||||
const closeOutput = vi.fn().mockResolvedValue(true);
|
||||
(
|
||||
channel as unknown as {
|
||||
interactionPresenter: {
|
||||
appendOutput: (_segment: unknown, chunk: string) => void;
|
||||
closeOutput: typeof closeOutput;
|
||||
};
|
||||
}
|
||||
).interactionPresenter = {
|
||||
appendOutput: (_segment, chunk) => projected.push(chunk),
|
||||
closeOutput,
|
||||
};
|
||||
const first = segment('segment-1');
|
||||
|
||||
getChunkHook(channel)('cid123', 'before\n[FI', 'session-1', first);
|
||||
await getOutputSegmentEndHook(channel)(
|
||||
'cid123',
|
||||
'session-1',
|
||||
first,
|
||||
'response_boundary',
|
||||
);
|
||||
// The base mints a fresh segment UUID after closeOutputSegment, but the
|
||||
// same run continues.
|
||||
const next = { ...first, segmentId: 'segment-2' };
|
||||
getChunkHook(channel)(
|
||||
'cid123',
|
||||
'LE: /workspace/secret.txt]\nafter',
|
||||
'session-1',
|
||||
next,
|
||||
);
|
||||
|
||||
expect(projected.join('')).toBe('before\n\nafter');
|
||||
expect(projected.join('')).not.toContain('secret.txt');
|
||||
});
|
||||
|
||||
it('delivers the line after a marker line ending exactly on a block boundary', async () => {
|
||||
const channel = createChannel({ blockStreaming: 'on' });
|
||||
seedWebhook(channel, 'cid123');
|
||||
getPromptHook(channel, 'onPromptStart')('cid123', 'session-1');
|
||||
const fetchSpy = vi
|
||||
.spyOn(globalThis, 'fetch')
|
||||
.mockResolvedValue(new Response('{}'));
|
||||
const send = getResponseHook(channel);
|
||||
const streamer = new BlockStreamer({
|
||||
minChars: 20,
|
||||
maxChars: 1000,
|
||||
idleMs: 0,
|
||||
send: (text) => send('cid123', text, 'session-1'),
|
||||
});
|
||||
|
||||
streamer.push(
|
||||
'[FILE: /workspace/report.txt]\n\nThe answer is 42.\nSecond line',
|
||||
);
|
||||
await streamer.flush();
|
||||
|
||||
const bodies = JSON.stringify(fetchSpy.mock.calls);
|
||||
expect(bodies).toContain('The answer is 42.');
|
||||
expect(bodies).toContain('File delivery unavailable');
|
||||
expect(bodies).not.toContain('/workspace/report.txt');
|
||||
});
|
||||
|
||||
it('drops the block projector when its session dies', async () => {
|
||||
const channel = createChannel({ blockStreaming: 'on' });
|
||||
seedWebhook(channel, 'cid123');
|
||||
getPromptHook(channel, 'onPromptStart')('cid123', 'session-1');
|
||||
vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response('{}'));
|
||||
|
||||
await getResponseHook(channel)('cid123', 'Answer text', 'session-1');
|
||||
const projectors = (
|
||||
channel as unknown as { blockFileProjectors: Map<string, unknown> }
|
||||
).blockFileProjectors;
|
||||
expect(projectors.size).toBe(1);
|
||||
|
||||
channel.onSessionDied('session-1');
|
||||
expect(projectors.size).toBe(0);
|
||||
});
|
||||
|
||||
it("keeps other sessions' status projectors when one session dies", () => {
|
||||
const channel = createChannel();
|
||||
const sessionOne = segment('segment-a');
|
||||
const sessionTwo = {
|
||||
...segment('segment-b'),
|
||||
sessionId: 'session-2',
|
||||
runId: 'run-2',
|
||||
};
|
||||
getChunkHook(channel)('cid123', 'chunk a', 'session-1', sessionOne);
|
||||
getChunkHook(channel)('cid123', 'chunk b', 'session-2', sessionTwo);
|
||||
const projectors = (
|
||||
channel as unknown as {
|
||||
fileProjectors: Map<string, { sessionId: string }>;
|
||||
}
|
||||
).fileProjectors;
|
||||
expect(projectors.size).toBe(2);
|
||||
|
||||
channel.onSessionDied('session-1');
|
||||
expect(projectors.size).toBe(1);
|
||||
expect([...projectors.values()][0]!.sessionId).toBe('session-2');
|
||||
});
|
||||
|
||||
it('drops the status projector on the terminal lifecycle event', () => {
|
||||
const channel = createChannel();
|
||||
getChunkHook(channel)('cid123', 'streamed', 'session-1', segment());
|
||||
const projectors = (
|
||||
channel as unknown as { fileProjectors: Map<string, unknown> }
|
||||
).fileProjectors;
|
||||
expect(projectors.size).toBe(1);
|
||||
|
||||
getLifecycleHook(channel)({
|
||||
type: 'completed',
|
||||
channelName: 'dingtalk',
|
||||
chatId: 'cid123',
|
||||
sessionId: 'session-1',
|
||||
runId: 'run-1',
|
||||
identity: { id: 'channel:dingtalk', displayName: 'dingtalk' },
|
||||
memoryScope: { namespace: 'channel:dingtalk', mode: 'metadata-only' },
|
||||
});
|
||||
expect(projectors.size).toBe(0);
|
||||
});
|
||||
|
||||
it('keeps file projection when rebuilding image-replaced text', async () => {
|
||||
const channel = createChannel();
|
||||
const prepare = (
|
||||
channel as unknown as {
|
||||
prepareOutgoingText(text: string): Promise<string>;
|
||||
}
|
||||
).prepareOutgoingText.bind(channel);
|
||||
|
||||
const out = await prepare(
|
||||
'before\n[FILE: /workspace/report.txt]\n[IMAGE: /workspace/missing.png]\nafter',
|
||||
);
|
||||
|
||||
expect(out).toContain('[File delivery unavailable]');
|
||||
expect(out).not.toContain('[FILE:');
|
||||
expect(out).not.toContain('/workspace/report.txt');
|
||||
expect(out).toContain('[Image delivery failed: missing.png]');
|
||||
});
|
||||
});
|
||||
|
||||
describe('DingtalkChannel reply delivery timeout', () => {
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
it('bounds the webhook POST with an abort timeout', async () => {
|
||||
const channel = createChannel();
|
||||
seedWebhook(channel, 'cid123');
|
||||
const controller = new AbortController();
|
||||
const timeoutSpy = vi
|
||||
.spyOn(AbortSignal, 'timeout')
|
||||
.mockReturnValue(controller.signal);
|
||||
vi.spyOn(globalThis, 'fetch').mockImplementation((_url, init) => {
|
||||
const signal = (init as RequestInit | undefined)?.signal;
|
||||
return new Promise<Response>((_resolve, reject) => {
|
||||
signal?.addEventListener('abort', () => reject(signal.reason));
|
||||
});
|
||||
});
|
||||
|
||||
const send = channel.sendMessage('cid123', 'hello');
|
||||
const outcome = await Promise.race([
|
||||
send.then(
|
||||
() => 'resolved',
|
||||
(err: unknown) =>
|
||||
`rejected: ${err instanceof Error ? err.message : String(err)}`,
|
||||
),
|
||||
new Promise<string>((resolve) =>
|
||||
setTimeout(() => resolve('pending'), 20),
|
||||
),
|
||||
]);
|
||||
expect(outcome).toBe('pending');
|
||||
expect(timeoutSpy).toHaveBeenCalledWith(15_000);
|
||||
|
||||
controller.abort(new Error('reply fetch timed out'));
|
||||
await expect(send).rejects.toThrow('reply fetch timed out');
|
||||
});
|
||||
});
|
||||
|
||||
describe('DingtalkChannel proactive send', () => {
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
|
||||
@@ -27,6 +27,11 @@ import {
|
||||
replaceImageMarkers,
|
||||
uploadDingTalkImage,
|
||||
} from './outbound-image.js';
|
||||
import {
|
||||
OutboundFileProjector,
|
||||
projectFileText,
|
||||
withFileUnavailableNotice,
|
||||
} from './outbound-file.js';
|
||||
import {
|
||||
DingtalkConnectionManager,
|
||||
type DingtalkManagedSocket,
|
||||
@@ -584,6 +589,7 @@ const DIRECT_MSG_API =
|
||||
const PROACTIVE_MSG_KEY = 'sampleMarkdown'; // DingTalk's built-in {title, text} markdown template key
|
||||
const TOKEN_API = 'https://oapi.dingtalk.com/gettoken';
|
||||
const PROACTIVE_FETCH_TIMEOUT_MS = 15_000;
|
||||
const REPLY_FETCH_TIMEOUT_MS = 15_000;
|
||||
// Extensions for generated media store names, keyed by the download's mime
|
||||
// type. The agent reads stored media via `read_file`, whose type detection is
|
||||
// extension-first: an extensionless name falls through to the binary content
|
||||
@@ -701,6 +707,24 @@ export class DingtalkChannel extends ChannelBase {
|
||||
private readonly inboundCardOwners = new Map<string, CardRunCorrelation>();
|
||||
private readonly cardRunBySession = new Map<string, string>();
|
||||
private readonly cardRuns = new Map<string, CardRunCorrelation>();
|
||||
// Keyed by runId, not segmentId: a mid-turn segment reset (response
|
||||
// boundary, input requested) mints a fresh segment UUID but the projection
|
||||
// state must survive it, or a marker split across the reset leaks.
|
||||
private readonly fileProjectors = new Map<
|
||||
string,
|
||||
{ sessionId: string; projector: OutboundFileProjector }
|
||||
>();
|
||||
private readonly blockFileProjectors = new Map<
|
||||
string,
|
||||
{ projector: OutboundFileProjector; reportedMarkers: number }
|
||||
>();
|
||||
// Sessions armed for block projection by onPromptStart and disarmed when
|
||||
// the turn settles (or the session dies). A block send that finds NO
|
||||
// projector state is only legitimate as a turn's FIRST block, which always
|
||||
// lands while armed: late sends from an evicted (/clear) or dead session
|
||||
// must be dropped, because recreating state would post the tail of a
|
||||
// force-split [FILE: ...] marker verbatim.
|
||||
private readonly blockProjectionArmed = new Set<string>();
|
||||
|
||||
constructor(
|
||||
name: string,
|
||||
@@ -1045,9 +1069,35 @@ export class DingtalkChannel extends ChannelBase {
|
||||
return isGroup && !conversationId;
|
||||
}
|
||||
|
||||
private async prepareOutgoingText(text: string): Promise<string> {
|
||||
const markers = findImageMarkers(text);
|
||||
if (markers.length === 0) return text;
|
||||
private projectOutgoingFileText(
|
||||
text: string,
|
||||
streamed?: OutboundFileProjector,
|
||||
): string {
|
||||
const projection = projectFileText(text);
|
||||
// Markers are counted when their opening bytes arrive, so the streamed
|
||||
// count also fails closed when the final text no longer carries a marker
|
||||
// the stream already delivered. Whole-turn hash comparison is NOT
|
||||
// viable: the bridges return only post-last-boundary chunks as the final
|
||||
// text, so any routine multi-tool turn would diverge by construction.
|
||||
const streamedMarkers = streamed ? streamed.result('').markerCount : 0;
|
||||
if (projection.markerCount === 0 && streamedMarkers === 0) {
|
||||
return projection.text;
|
||||
}
|
||||
// Counts only — never paths — so a redaction event stays debuggable
|
||||
// without leaking what was redacted.
|
||||
process.stderr.write(
|
||||
`[DingTalk:${this.name}] file markers redacted (final=${projection.markerCount}, streamed=${streamedMarkers})\n`,
|
||||
);
|
||||
return withFileUnavailableNotice(projection.text);
|
||||
}
|
||||
|
||||
private async prepareOutgoingText(
|
||||
text: string,
|
||||
streamed?: OutboundFileProjector,
|
||||
): Promise<string> {
|
||||
const fileSafeText = this.projectOutgoingFileText(text, streamed);
|
||||
const markers = findImageMarkers(fileSafeText);
|
||||
if (markers.length === 0) return fileSafeText;
|
||||
|
||||
const replacements: string[] = [];
|
||||
for (const marker of markers) {
|
||||
@@ -1095,7 +1145,7 @@ export class DingtalkChannel extends ChannelBase {
|
||||
}
|
||||
}
|
||||
|
||||
return replaceImageMarkers(text, markers, replacements);
|
||||
return replaceImageMarkers(fileSafeText, markers, replacements);
|
||||
}
|
||||
|
||||
private async sendReply(
|
||||
@@ -1129,11 +1179,23 @@ export class DingtalkChannel extends ChannelBase {
|
||||
...(isMention ? { at: { atUserIds: [atUserId] } } : {}),
|
||||
};
|
||||
|
||||
const resp = await fetch(webhook, {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify(body),
|
||||
});
|
||||
let resp: Response;
|
||||
try {
|
||||
resp = await fetch(webhook, {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify(body),
|
||||
signal: AbortSignal.timeout(REPLY_FETCH_TIMEOUT_MS),
|
||||
});
|
||||
} catch (err) {
|
||||
process.stderr.write(
|
||||
`[DingTalk:${this.name}] sendMessage failed: ${sanitizeLogText(
|
||||
err instanceof Error ? err.message : String(err),
|
||||
300,
|
||||
)}\n`,
|
||||
);
|
||||
throw err;
|
||||
}
|
||||
|
||||
if (isMention && process.env['QWEN_CHANNEL_DEBUG_MENTIONS'] === '1') {
|
||||
const payload = (await resp
|
||||
@@ -1551,6 +1613,11 @@ export class DingtalkChannel extends ChannelBase {
|
||||
|
||||
/** Recall reactions left behind when a session dies without terminal lifecycle events. */
|
||||
override onSessionDied(sessionId: string): void {
|
||||
this.blockProjectionArmed.delete(sessionId);
|
||||
this.blockFileProjectors.delete(sessionId);
|
||||
for (const [runId, state] of this.fileProjectors) {
|
||||
if (state.sessionId === sessionId) this.fileProjectors.delete(runId);
|
||||
}
|
||||
const bufferedTargets = this.bufferedMentionTargetsBySession.get(sessionId);
|
||||
if (bufferedTargets) {
|
||||
this.bufferedMentionTargetsBySession.delete(sessionId);
|
||||
@@ -1609,6 +1676,7 @@ export class DingtalkChannel extends ChannelBase {
|
||||
if (event.messageId) this.mentionTargets.delete(event.messageId);
|
||||
this.stopReaction(event.chatId, event.messageId, event.sessionId);
|
||||
if (event.runId) {
|
||||
this.deleteFileProjectorsForRun(event.runId);
|
||||
if (event.type === 'failed') {
|
||||
this.interactionPresenter?.terminalizeRun(
|
||||
event.runId,
|
||||
@@ -1679,6 +1747,7 @@ export class DingtalkChannel extends ChannelBase {
|
||||
sessionId: string,
|
||||
messageId?: string,
|
||||
): void {
|
||||
this.blockProjectionArmed.add(sessionId);
|
||||
if (messageId) {
|
||||
this.bufferedMentionTargets.delete(messageId);
|
||||
this.untrackBufferedMentionTarget(sessionId, messageId);
|
||||
@@ -1751,20 +1820,103 @@ export class DingtalkChannel extends ChannelBase {
|
||||
sessionId: string,
|
||||
messageId?: string,
|
||||
): void {
|
||||
this.settleBlockFileProjector(chatId, sessionId);
|
||||
this.sessionMentionTargets.delete(sessionId);
|
||||
this.stopReaction(chatId, messageId, sessionId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Turn end is the only point where the block projector's held state can be
|
||||
* settled: ChannelBase drains the turn's queued block sends before calling
|
||||
* onPromptEnd, so everything already appended belongs to this turn. Flush
|
||||
* the held candidate bytes (a trailing `[FILE:` prefix the stream never
|
||||
* completed) before deleting the entry — a later delete-without-settle
|
||||
* would silently drop them from the delivered answer. Also disarm the
|
||||
* session: /clear eviction and session death settle WITHOUT draining the
|
||||
* turn's send chain, and any block that lands afterwards must be dropped
|
||||
* rather than recreating projector state.
|
||||
*/
|
||||
private settleBlockFileProjector(chatId: string, sessionId: string): void {
|
||||
this.blockProjectionArmed.delete(sessionId);
|
||||
const state = this.blockFileProjectors.get(sessionId);
|
||||
if (!state) return;
|
||||
this.blockFileProjectors.delete(sessionId);
|
||||
const tail = state.projector.complete();
|
||||
if (!tail.trim()) return;
|
||||
// Deliberately fire-and-forget: complete() can only return a strict
|
||||
// prefix of '[FILE:' (at most 5 chars, no path bytes), so the worst case
|
||||
// is a stray fragment landing out of order with the next turn — not a
|
||||
// leak — and blocking settle on a delivery that may hang is worse.
|
||||
void this.sendReply(chatId, tail).catch((err) => {
|
||||
process.stderr.write(
|
||||
`[DingTalk:${this.name}] projector tail delivery failed for session ${sessionId}: ${err instanceof Error ? err.message : String(err)}\n`,
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Out-of-turn one-shot sends (background responses) must not flow through
|
||||
* the session's block-streaming projector: a second sender interleaving with
|
||||
* mid-projection state can swallow the send or split a held marker.
|
||||
*/
|
||||
protected override async deliverBackgroundReply(
|
||||
chatId: string,
|
||||
text: string,
|
||||
sessionId: string,
|
||||
): Promise<void> {
|
||||
if (this.config.blockStreaming !== 'on') {
|
||||
return super.deliverBackgroundReply(chatId, text, sessionId);
|
||||
}
|
||||
await this.sendReply(chatId, text);
|
||||
}
|
||||
|
||||
protected override async sendResponseMessage(
|
||||
chatId: string,
|
||||
text: string,
|
||||
sessionId: string,
|
||||
): Promise<void> {
|
||||
const atUserId = this.atSender
|
||||
? this.sessionMentionTargets.get(sessionId)
|
||||
: undefined;
|
||||
let outgoingText = text;
|
||||
let consumesMention = true;
|
||||
if (this.config.blockStreaming === 'on') {
|
||||
const projected = this.projectBlockStreamChunk(text, sessionId);
|
||||
if (!projected.text.trim()) return;
|
||||
outgoingText = projected.text;
|
||||
// A notice-only block must not consume the prompt's mention target:
|
||||
// the @mention belongs to the block carrying the actual answer.
|
||||
consumesMention = projected.hasContent;
|
||||
}
|
||||
const atUserId =
|
||||
consumesMention && this.atSender
|
||||
? this.sessionMentionTargets.get(sessionId)
|
||||
: undefined;
|
||||
if (atUserId) this.sessionMentionTargets.delete(sessionId);
|
||||
await this.sendReply(chatId, text, atUserId);
|
||||
await this.sendReply(chatId, outgoingText, atUserId);
|
||||
}
|
||||
|
||||
private projectBlockStreamChunk(
|
||||
text: string,
|
||||
sessionId: string,
|
||||
): { text: string; hasContent: boolean } {
|
||||
let state = this.blockFileProjectors.get(sessionId);
|
||||
if (!state) {
|
||||
if (!this.blockProjectionArmed.has(sessionId)) {
|
||||
return { text: '', hasContent: false };
|
||||
}
|
||||
state = { projector: new OutboundFileProjector(), reportedMarkers: 0 };
|
||||
this.blockFileProjectors.set(sessionId, state);
|
||||
}
|
||||
const safe = state.projector.append(text);
|
||||
const hasContent = safe.trim().length > 0;
|
||||
const result = state.projector.result(safe);
|
||||
let outgoingText = safe;
|
||||
if (result.markerCount > state.reportedMarkers) {
|
||||
outgoingText = withFileUnavailableNotice(safe);
|
||||
state.reportedMarkers = result.markerCount;
|
||||
process.stderr.write(
|
||||
`[DingTalk:${this.name}] file markers redacted in block stream (session ${sessionId}, markers=${result.markerCount})\n`,
|
||||
);
|
||||
}
|
||||
return { text: outgoingText, hasContent };
|
||||
}
|
||||
|
||||
private async sendFallbackReply(
|
||||
@@ -1786,8 +1938,12 @@ export class DingtalkChannel extends ChannelBase {
|
||||
sessionId: string,
|
||||
segment?: ChannelOutputSegmentContext,
|
||||
): Promise<void> {
|
||||
const streamed = segment
|
||||
? this.fileProjectors.get(segment.runId)?.projector
|
||||
: undefined;
|
||||
if (segment) this.fileProjectors.delete(segment.runId);
|
||||
const outgoingText = await this.prepareOutgoingText(text, streamed);
|
||||
if (segment && this.interactionPresenter) {
|
||||
const outgoingText = await this.prepareOutgoingText(text);
|
||||
if (
|
||||
await this.interactionPresenter.closeOutput(
|
||||
segment.segmentId,
|
||||
@@ -1799,7 +1955,7 @@ export class DingtalkChannel extends ChannelBase {
|
||||
return;
|
||||
}
|
||||
}
|
||||
await this.sendResponseMessage(chatId, text, sessionId);
|
||||
await this.sendResponseMessage(chatId, outgoingText, sessionId);
|
||||
}
|
||||
|
||||
protected override onOutputSegmentEnd(
|
||||
@@ -1808,6 +1964,13 @@ export class DingtalkChannel extends ChannelBase {
|
||||
segment: ChannelOutputSegmentContext,
|
||||
reason: ChannelOutputSegmentEndReason,
|
||||
): void | Promise<void> {
|
||||
if (
|
||||
reason === 'completed' ||
|
||||
reason === 'failed' ||
|
||||
reason === 'cancelled'
|
||||
) {
|
||||
this.fileProjectors.delete(segment.runId);
|
||||
}
|
||||
if (!this.interactionPresenter) return;
|
||||
return this.interactionPresenter
|
||||
.closeOutput(segment.segmentId, '', reason, segment)
|
||||
@@ -1820,7 +1983,21 @@ export class DingtalkChannel extends ChannelBase {
|
||||
_sessionId: string,
|
||||
segment?: ChannelOutputSegmentContext,
|
||||
): void {
|
||||
if (segment) this.interactionPresenter?.appendOutput(segment, chunk);
|
||||
if (!segment) return;
|
||||
let state = this.fileProjectors.get(segment.runId);
|
||||
if (!state) {
|
||||
state = {
|
||||
sessionId: segment.sessionId,
|
||||
projector: new OutboundFileProjector(),
|
||||
};
|
||||
this.fileProjectors.set(segment.runId, state);
|
||||
}
|
||||
const safe = state.projector.append(chunk);
|
||||
if (safe) this.interactionPresenter?.appendOutput(segment, safe);
|
||||
}
|
||||
|
||||
private deleteFileProjectorsForRun(runId: string): void {
|
||||
this.fileProjectors.delete(runId);
|
||||
}
|
||||
|
||||
protected override async presentUserInputRequest(
|
||||
|
||||
@@ -0,0 +1,84 @@
|
||||
import { describe, expect, it } from 'vitest';
|
||||
import {
|
||||
MAX_FILES_PER_RESPONSE,
|
||||
OutboundFileProjector,
|
||||
projectFileText,
|
||||
} from './outbound-file.js';
|
||||
|
||||
describe('OutboundFileProjector', () => {
|
||||
it('keeps every split of the reserved opening path-free', () => {
|
||||
const input = 'before\n[FILE: /workspace/report.txt]\nafter';
|
||||
for (let split = 0; split <= '[FILE:'.length; split++) {
|
||||
const projector = new OutboundFileProjector();
|
||||
const opening = input.indexOf('[FILE:');
|
||||
const chunks = [
|
||||
input.slice(0, opening + split),
|
||||
input.slice(opening + split),
|
||||
];
|
||||
const safeChunks = chunks.map((chunk) => projector.append(chunk));
|
||||
const safe = safeChunks.join('') + projector.complete();
|
||||
expect(safeChunks[0]).not.toContain('/workspace/report.txt');
|
||||
expect(safeChunks[1]).not.toContain('/workspace/report.txt');
|
||||
expect(projector.result(safe)).toMatchObject({
|
||||
text: 'before\n\nafter',
|
||||
paths: ['/workspace/report.txt'],
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
it.each([
|
||||
{
|
||||
input: '[FILE: /tmp/a.txt]',
|
||||
text: '',
|
||||
paths: ['/tmp/a.txt'],
|
||||
invalidMarkers: 0,
|
||||
},
|
||||
{
|
||||
input: 'prefix [FILE: /tmp/a.txt] suffix\nnext',
|
||||
text: 'prefix \nnext',
|
||||
paths: [],
|
||||
invalidMarkers: 1,
|
||||
},
|
||||
{
|
||||
input: '[FILE:/tmp/a.txt]\nnext',
|
||||
text: '\nnext',
|
||||
paths: [],
|
||||
invalidMarkers: 1,
|
||||
},
|
||||
{
|
||||
input: '[FILE: /tmp/a.txt',
|
||||
text: '',
|
||||
paths: [],
|
||||
invalidMarkers: 1,
|
||||
},
|
||||
])('projects $input without repairing it', ({ input, ...expected }) => {
|
||||
expect(projectFileText(input)).toMatchObject(expected);
|
||||
});
|
||||
|
||||
it('does not rescan text joined by a redaction', () => {
|
||||
expect(projectFileText('[FI[FILE: /tmp/inner]\nLE: /tmp/outer]\n')).toEqual(
|
||||
{
|
||||
text: '[FI\nLE: /tmp/outer]\n',
|
||||
paths: [],
|
||||
invalidMarkers: 1,
|
||||
excessMarkers: 0,
|
||||
markerCount: 1,
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
it('bounds accepted paths and rejects oversized reserved lines', () => {
|
||||
const markers = Array.from(
|
||||
{ length: MAX_FILES_PER_RESPONSE + 2 },
|
||||
(_, index) => `[FILE: /tmp/${index}.txt]`,
|
||||
).join('\n');
|
||||
const projected = projectFileText(
|
||||
`${markers}\n[FILE: /${'x'.repeat(5000)}]`,
|
||||
);
|
||||
expect(projected.paths).toHaveLength(MAX_FILES_PER_RESPONSE);
|
||||
expect(projected.excessMarkers).toBe(2);
|
||||
expect(projected.invalidMarkers).toBe(1);
|
||||
expect(projected.text).not.toContain('/tmp/');
|
||||
expect(projected.text).not.toContain('x'.repeat(100));
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,121 @@
|
||||
const FILE_OPENING = '[FILE:';
|
||||
const MAX_FILE_PATH_CHARS = 4096;
|
||||
export const MAX_FILES_PER_RESPONSE = 5;
|
||||
export const FILE_UNAVAILABLE_NOTICE = '[File delivery unavailable]';
|
||||
|
||||
export interface FileProjection {
|
||||
text: string;
|
||||
paths: string[];
|
||||
invalidMarkers: number;
|
||||
excessMarkers: number;
|
||||
markerCount: number;
|
||||
}
|
||||
|
||||
export class OutboundFileProjector {
|
||||
private candidate = '';
|
||||
private reserved = '';
|
||||
private reservedAtLineStart = false;
|
||||
private reservedTooLong = false;
|
||||
private atLineStart = true;
|
||||
private readonly paths: string[] = [];
|
||||
private invalidMarkers = 0;
|
||||
private excessMarkers = 0;
|
||||
private markerCount = 0;
|
||||
|
||||
append(chunk: string): string {
|
||||
let safe = '';
|
||||
for (const char of chunk) {
|
||||
if (this.reserved) {
|
||||
if (char === '\n') {
|
||||
this.finishReservedLine();
|
||||
safe += '\n';
|
||||
this.atLineStart = true;
|
||||
} else if (this.reserved.length <= MAX_FILE_PATH_CHARS + 9) {
|
||||
this.reserved += char;
|
||||
} else {
|
||||
this.reservedTooLong = true;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (char === '\n') {
|
||||
safe += `${this.candidate}\n`;
|
||||
this.candidate = '';
|
||||
this.atLineStart = true;
|
||||
continue;
|
||||
}
|
||||
this.candidate += char;
|
||||
while (this.candidate && !FILE_OPENING.startsWith(this.candidate)) {
|
||||
safe += this.candidate[0];
|
||||
this.candidate = this.candidate.slice(1);
|
||||
this.atLineStart = false;
|
||||
}
|
||||
if (this.candidate === FILE_OPENING) {
|
||||
this.reserved = this.candidate;
|
||||
this.reservedAtLineStart = this.atLineStart;
|
||||
this.candidate = '';
|
||||
this.markerCount++;
|
||||
}
|
||||
}
|
||||
return safe;
|
||||
}
|
||||
|
||||
complete(): string {
|
||||
if (!this.reserved) {
|
||||
const safe = this.candidate;
|
||||
this.candidate = '';
|
||||
return safe;
|
||||
}
|
||||
this.finishReservedLine();
|
||||
return '';
|
||||
}
|
||||
|
||||
result(text: string): FileProjection {
|
||||
return {
|
||||
text,
|
||||
paths: [...this.paths],
|
||||
invalidMarkers: this.invalidMarkers,
|
||||
excessMarkers: this.excessMarkers,
|
||||
markerCount: this.markerCount,
|
||||
};
|
||||
}
|
||||
|
||||
private finishReservedLine(): void {
|
||||
const match = /^\[FILE: ([^\]\r\n]+)\]\r?$/u.exec(this.reserved);
|
||||
const path = match?.[1];
|
||||
if (
|
||||
!this.reservedTooLong &&
|
||||
this.reservedAtLineStart &&
|
||||
path &&
|
||||
path.length <= MAX_FILE_PATH_CHARS &&
|
||||
path === path.trim()
|
||||
) {
|
||||
if (this.paths.length < MAX_FILES_PER_RESPONSE) {
|
||||
this.paths.push(path);
|
||||
} else {
|
||||
this.excessMarkers++;
|
||||
}
|
||||
} else {
|
||||
this.invalidMarkers++;
|
||||
}
|
||||
this.reserved = '';
|
||||
this.reservedAtLineStart = false;
|
||||
this.reservedTooLong = false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Redacts [FILE: ...] markers from one complete outbound text. Detection
|
||||
* deliberately scans the RAW text, unlike the image path's maskCode scan:
|
||||
* FILE performs no upload, so a FILE-shaped example inside a code fence is
|
||||
* over-redacted rather than risking a leaked path — security over fidelity.
|
||||
*/
|
||||
export function projectFileText(text: string): FileProjection {
|
||||
const projector = new OutboundFileProjector();
|
||||
const safe = projector.append(text) + projector.complete();
|
||||
return projector.result(safe);
|
||||
}
|
||||
|
||||
export function withFileUnavailableNotice(text: string): string {
|
||||
const safe = text.trimEnd();
|
||||
return `${safe}${safe ? '\n' : ''}${FILE_UNAVAILABLE_NOTICE}`;
|
||||
}
|
||||
Reference in new issue
Block a user