From 04d43e3b1932ee1b6282535230260b6404702493 Mon Sep 17 00:00:00 2001 From: ZiyaZhang <199893125+ZiyaZhang@users.noreply.github.com> Date: Sun, 2 Aug 2026 05:58:31 -0700 Subject: [PATCH] feat(llm-replay): resolve {{fromRequest:...}} placeholders against the live request A scripted sidecar cannot know values minted at run time, so terminal goal updates (which must echo the random goal id) were previously un-scriptable. Placeholders in scripted entries now resolve against the request corpus at stream time: last match wins, capture group 1 or the whole match substitutes, and unmatched/invalid/unterminated patterns fail loud. --- packages/support/llm-replay/README.i18n.yaml | 4 +- packages/support/llm-replay/README.md | 4 +- packages/support/llm-replay/README.zh.md | 4 +- packages/support/llm-replay/src/index.ts | 86 ++++++++++++++++++- .../llm-replay/tests/llm-replay.spec.ts | 71 ++++++++++++++- 5 files changed, 163 insertions(+), 6 deletions(-) diff --git a/packages/support/llm-replay/README.i18n.yaml b/packages/support/llm-replay/README.i18n.yaml index e4f6711ea7..b6a31ebb99 100644 --- a/packages/support/llm-replay/README.i18n.yaml +++ b/packages/support/llm-replay/README.i18n.yaml @@ -2,5 +2,5 @@ # side as of the last confirmed-consistent state. Both languages carry equal authority; # after editing either side, bring the other along and re-record with: # pnpm run verify-translation-pairing --write packages/support/llm-replay/README.md -README.md: 0deb6e76b29d40483b754ac01c98ee0e01bfcbe8 -README.zh.md: 7720e2d1bc6eb7bc5c89d5c1708767a54a7b0080 +README.md: ea52525ee85aae58006c852afe93291ea70807d5 +README.zh.md: e8c0ec225df29fc6f5493776d75bc7f9e3e078de diff --git a/packages/support/llm-replay/README.md b/packages/support/llm-replay/README.md index 0deb6e76b2..ea52525ee8 100644 --- a/packages/support/llm-replay/README.md +++ b/packages/support/llm-replay/README.md @@ -12,6 +12,8 @@ The fixture IS the persisted session log (`/session.jsonl`). Its `assi Two failure modes are not reconstructable from `assistant/chunk` alone — a pure throw before any chunk (e.g. an HTTP 401, where the log holds only a `turn/end {error}` and no chunks) and a cancel/hang (timing, not chunk content). A scenario that needs those supplies an optional sidecar (`/replay.override.json`) that either replaces the derived script (a bare `ReplayEntry[]`) or augments it (`{ patches: [{ at, entry }] }`: keep every JSONL-derived call and swap the named 0-based call indexes; `at` equal to the derived length appends the retry attempt after an injected transient throw). Patch indexes must be unique. The override document, each patch and entry, and every chunk discriminant are validated when the file loads. A `hang` entry may name `readyFile`; replay writes that empty marker after its prefix chunks reach the loop and before it waits for cancellation, so an external driver can cancel deterministically without observing a presentation update. +A scripted string may embed `{{fromRequest:}}` to fill a value no static sidecar can know — for example a randomly minted goal id the model must echo back into `update_goal`. At stream time every placeholder resolves against the live request: the corpus is every string leaf of the request messages joined by newlines, the pattern's LAST corpus match wins, and its first capture group (or the whole match without one) substitutes in place. A pattern that matches nothing, an invalid pattern, and an unterminated placeholder each fail loud; the first `}}` ends the placeholder, so patterns cannot contain `}}`. + ## Nested agents: per-session keying A scenario where a parent agent delegates to in-process subagents records more than one log: the parent (`session.jsonl`) plus one per child (`session.1.jsonl`, …). Each agent runs as its own `Session` on the same context, so replay must serve each one its own script. @@ -55,7 +57,7 @@ Replay keys every call by its calling session id (`GenerateOptions.sessionId`, s - `installLlmReplay(ctx, config)` — install the configured replay adapter or catch-all `llm/stream` listener; returns a `ReplayHandle` (`dispose()` for HMR safety plus `assertConsumed()`, the teardown check that every recorded script bound to a live session and every bound cursor drained — turning a scenario that silently drove fewer model calls than recorded into a crisp diagnostic). Use this in tests to drive replay without the Loader or env vars. - `loadSessionScripts(config)` — resolve the ordered `SessionScript[]` (primary + children) for a scenario, ready to bind to live sessions in first-call order. - `loadReplayScript(config)` — resolve the `ReplayEntry[]` for the primary session only (validated sidecar replacement/patches if present, else derived from the JSONL; fail-loud if the fixture is missing). -- `deriveReplayScript(events)` / `parseSessionLog(text)` / `parseSessionHeader(text)` — the pure helpers that turn a recorded session log into a script and read its header `id`/`createdAt`. A derived group must end in a `finish` chunk; a group without one is the fingerprint of a thrown `stream()` and must instead be expressed via an override sidecar. +- `deriveReplayScript(events)` / `parseSessionLog(text)` / `parseSessionHeader(text)` / `resolveScriptedEntry(entry, messages)` — the pure helpers that turn a recorded session log into a script, read its header `id`/`createdAt`, and resolve `{{fromRequest:...}}` placeholders against one live request. A derived group must end in a `finish` chunk; a group without one is the fingerprint of a thrown `stream()` and must instead be expressed via an override sidecar. - Types `ReplayEntry` / `ReplayOverrideDoc` / `ReplayOverridePatch` / `SessionScript` / `ReplayConfig` / `ReplayProviderConfig` / `ReplayModelConfig` / `ReplayHandle` / `Config`. ## Plugin export shape diff --git a/packages/support/llm-replay/README.zh.md b/packages/support/llm-replay/README.zh.md index 7720e2d1bc..e8c0ec225d 100644 --- a/packages/support/llm-replay/README.zh.md +++ b/packages/support/llm-replay/README.zh.md @@ -12,6 +12,8 @@ fixture 就是持久化的会话日志（`/session.jsonl`）。其 `as 有两种失败模式无法仅根据 `assistant/chunk` 重建：在产生任何分片前直接抛出异常（例如 HTTP 401，此时日志只有 `turn/end {error}` 而没有分片），以及取消或挂起（差异在时序，而非分片内容）。需要这些行为的场景可提供伴随文件（`/replay.override.json`）：它可以替换派生脚本（裸 `ReplayEntry[]`），也可以增补派生脚本（`{ patches: [{ at, entry }] }`：保留所有从 JSONL 派生的调用，只替换指定的从 0 开始计数的调用索引；当 `at` 等于派生长度时，则在注入瞬态异常后的重试位置追加一次调用）。补丁索引不得重复。文件加载时会校验覆写文档、每个补丁和条目，以及每个分片的判别标签。`hang` 条目可以指定 `readyFile`；当前缀分片到达循环后、开始等待取消前，回放会写入这个空标记，使外部驱动程序无需观察展示层更新即可确定性地取消。 +脚本字符串可以内嵌 `{{fromRequest:}}`，用来填入静态伴随文件不可能预知的值——例如模型必须原样回填到 `update_goal` 的随机生成 goal id。回放时每个占位符针对实时请求解析：语料是请求消息的所有字符串叶子按换行拼接的结果，取该模式在语料中的最后一次匹配，用其第一个捕获组（无捕获组时用整个匹配）原位替换。模式匹配不到内容、模式非法、占位符未闭合都会明确报错；第一个 `}}` 即结束占位符，因此模式本身不能包含 `}}`。 + ## 嵌套 agent：每会话键控 父 agent 委托给进程内 subagent（子 agent）的场景会记录多个日志：父会话使用 `session.jsonl`，每个子会话各使用一个日志（`session.1.jsonl` 等）。每个 agent 都在同一上下文中作为独立的 `Session` 运行，因此回放必须为每个 agent 提供各自的脚本。 @@ -55,7 +57,7 @@ fixture 就是持久化的会话日志（`/session.jsonl`）。其 `as - `installLlmReplay(ctx, config)`：安装已配置回放适配器或 catch-all `llm/stream` 监听器；返回 `ReplayHandle`（包含用于保证 HMR（热模块替换）安全的 `dispose()`，以及清理阶段执行的 `assertConsumed()` 检查；后者确保每个已记录脚本都绑定到实时会话，且每个已绑定游标都已耗尽，从而将场景静默驱动的模型调用少于记录数转换为明确诊断）。在测试中使用它，可以不通过 Loader 或 env var 驱动回放。 - `loadSessionScripts(config)`：解析场景的有序的 `SessionScript[]`（主会话 + 子会话），准备按首次调用顺序绑定到实时会话。 - `loadReplayScript(config)`：只解析主会话的 `ReplayEntry[]`（如果伴随文件存在，则使用经校验的替换或补丁；否则从 JSONL 派生；fixture 缺失时明确报错）。 -- `deriveReplayScript(events)` / `parseSessionLog(text)` / `parseSessionHeader(text)`：将已记录会话日志转换为脚本并读取其 header `id`/`createdAt` 的纯辅助工具。派生分组必须以 `finish` 分片结束；没有该分片的分组是 `stream()` 抛出异常的指纹，必须改用 override sidecar 表达。 +- `deriveReplayScript(events)` / `parseSessionLog(text)` / `parseSessionHeader(text)` / `resolveScriptedEntry(entry, messages)`：将已记录会话日志转换为脚本、读取其 header `id`/`createdAt`、并针对单次实时请求解析 `{{fromRequest:...}}` 占位符的纯辅助工具。派生分组必须以 `finish` 分片结束；没有该分片的分组是 `stream()` 抛出异常的指纹，必须改用 override sidecar 表达。 - 类型 `ReplayEntry` / `ReplayOverrideDoc` / `ReplayOverridePatch` / `SessionScript` / `ReplayConfig` / `ReplayProviderConfig` / `ReplayModelConfig` / `ReplayHandle` / `Config`。 ## 插件导出形态 diff --git a/packages/support/llm-replay/src/index.ts b/packages/support/llm-replay/src/index.ts index bfb9859415..70dae88446 100644 --- a/packages/support/llm-replay/src/index.ts +++ b/packages/support/llm-replay/src/index.ts @@ -241,6 +241,90 @@ const REPLAY_CHUNK_TYPES = new Set([ 'finish', ]) +const FROM_REQUEST_OPEN = '{{fromRequest:' +const FROM_REQUEST_CLOSE = '}}' + +/** Collect every string leaf of one JSON-shaped value, in traversal order. */ +function collectStrings(value: unknown, out: string[]): void { + if (typeof value === 'string') { + out.push(value) + return + } + if (Array.isArray(value)) { + for (const item of value) collectStrings(item, out) + return + } + if (value !== null && typeof value === 'object') { + for (const item of Object.values(value)) collectStrings(item, out) + } +} + +/** Resolve one placeholder pattern against the request corpus; the LAST match wins. */ +function resolveFromRequest(pattern: string, corpus: string): string { + let regex: RegExp + try { + regex = new RegExp(pattern, 'g') + } catch (error) { + // RegExp construction only throws SyntaxError; String() carries its message. + throw new Error(`llm-replay: fromRequest has an invalid pattern ${JSON.stringify(pattern)}: ${String(error)}`) + } + let last: RegExpExecArray | undefined + for (const match of corpus.matchAll(regex)) last = match + if (last === undefined) { + throw new Error(`llm-replay: fromRequest pattern ${JSON.stringify(pattern)} matched nothing in the request`) + } + return last[1] ?? last[0] +} + +/** Replace every `{{fromRequest:}}` occurrence in one scripted string. */ +function substituteString(text: string, corpus: string): string { + let result = '' + let cursor = 0 + while (true) { + const open = text.indexOf(FROM_REQUEST_OPEN, cursor) + if (open === -1) return result + text.slice(cursor) + const close = text.indexOf(FROM_REQUEST_CLOSE, open + FROM_REQUEST_OPEN.length) + if (close === -1) { + throw new Error(`llm-replay: fromRequest placeholder is unterminated in ${JSON.stringify(text)}`) + } + const pattern = text.slice(open + FROM_REQUEST_OPEN.length, close) + result += text.slice(cursor, open) + resolveFromRequest(pattern, corpus) + cursor = close + FROM_REQUEST_CLOSE.length + } +} + +/** Deep-copy one JSON-shaped value with scripted placeholders resolved. */ +function substituteValue(value: unknown, corpus: string): unknown { + if (typeof value === 'string') { + return value.includes(FROM_REQUEST_OPEN) ? substituteString(value, corpus) : value + } + if (Array.isArray(value)) return value.map(item => substituteValue(item, corpus)) + if (value !== null && typeof value === 'object') { + return Object.fromEntries(Object.entries(value).map(([key, item]) => [key, substituteValue(item, corpus)])) + } + return value +} + +/** + * Resolve every `{{fromRequest:}}` placeholder in one scripted entry + * against the live request. The corpus is every string leaf of the request + * messages joined by newlines; the pattern's LAST corpus match wins and its + * first capture group (or, without one, the whole match) substitutes in place. + * Scenario sidecars use this to script arguments no static file can know, + * such as a randomly minted goal id the model must echo back. A pattern that + * matches nothing, an invalid pattern, and an unterminated placeholder each + * fail loud. Patterns cannot contain `}}` — the first `}}` ends the placeholder. + * @param entry - the scripted entry about to replay. + * @param messages - the live request messages searched by the placeholders. + * @returns the entry itself when no placeholder appears, else a resolved deep copy. + */ +export function resolveScriptedEntry(entry: ReplayEntry, messages: GenerateOptions['messages']): ReplayEntry { + if (!JSON.stringify(entry).includes(FROM_REQUEST_OPEN)) return entry + const leaves: string[] = [] + collectStrings(messages, leaves) + return substituteValue(entry, leaves.join('\n')) as ReplayEntry +} + function isRecord(value: unknown): value is Record { return typeof value === 'object' && value !== null && !Array.isArray(value) } @@ -583,7 +667,7 @@ export function installLlmReplay(ctx: Context, config: ReplayConfig): ReplayHand + `but its script has only ${boundState.entries.length}; re-record the scenario`, ) } - yield* replayEntry(entry, options.signal, paceMs) + yield* replayEntry(resolveScriptedEntry(entry, options.messages), options.signal, paceMs) })() } const providers = config.providers ?? [] diff --git a/packages/support/llm-replay/tests/llm-replay.spec.ts b/packages/support/llm-replay/tests/llm-replay.spec.ts index d693c0571b..345aa22679 100644 --- a/packages/support/llm-replay/tests/llm-replay.spec.ts +++ b/packages/support/llm-replay/tests/llm-replay.spec.ts @@ -4,7 +4,7 @@ import { join } from 'node:path' import { afterEach, beforeEach, describe, expect, it } from 'vitest' import { Context } from 'cordis' import type { SessionEvent } from '@deepseek-ai/dsh-session' -import LlmService, { GenerateOptions, LlmAdapter, StreamChunk } from '@deepseek-ai/dsh-llm' +import LlmService, { CallId, createUserMessage, GenerateOptions, LlmAdapter, StreamChunk } from '@deepseek-ai/dsh-llm' import { type ReplayEntry, type SessionScript, @@ -17,6 +17,7 @@ import { name, parseSessionHeader, parseSessionLog, + resolveScriptedEntry, } from '../src/index.ts' /** @@ -310,6 +311,74 @@ describe('installLlmReplay (through the real LlmService)', () => { expect(await drain(ctx.llm.stream({ provider: 'm', model: 'm', messages: [] }))).toEqual(TEXT_CHUNKS) }) + describe('{{fromRequest:...}} substitution', () => { + const requestMessages = [createUserMessage({ + content: [{ type: 'text' as const, text: 'stale {"goal":{"id":"goal-old"}} then {"goal":{"id":"goal-42ab"}}' }], + source: { kind: 'user' as const }, + })] + + function scriptedCall(argumentsDelta: string): StreamChunk[] { + return [ + { type: 'block-start', index: 0, blockType: 'tool-call' }, + { type: 'tool-call-delta', index: 0, id: CallId('c1'), name: 'update_goal', argumentsDelta }, + { type: 'block-end', index: 0, block: { type: 'tool-call', id: CallId('c1'), name: 'update_goal', arguments: argumentsDelta } }, + { type: 'finish', reason: { kind: 'tool-calls' } }, + ] + } + + async function streamScripted(argumentsDelta: string): Promise { + writeLog(TEXT_CHUNKS) + const overrideFile = join(dir, 'replay.override.json') + writeFileSync(overrideFile, JSON.stringify([{ kind: 'chunks', chunks: scriptedCall(argumentsDelta) }]), 'utf8') + const ctx = new Context() + await ctx.plugin(LlmService) + installLlmReplay(ctx, { file, overrideFile }) + return drain(ctx.llm.stream({ provider: 'm', model: 'm', messages: requestMessages })) + } + + it('resolves the capture group from the LAST request match in every scripted string field', async () => { + const streamed = await streamScripted('{"goal_id":"{{fromRequest:"id":"(goal-[^"]+)"}}","revision":1}') + const delta = streamed.find(chunk => chunk.type === 'tool-call-delta') + expect(delta).toMatchObject({ argumentsDelta: '{"goal_id":"goal-42ab","revision":1}' }) + const end = streamed.find(chunk => chunk.type === 'block-end') + expect(end).toMatchObject({ block: { arguments: '{"goal_id":"goal-42ab","revision":1}' } }) + }) + + it('substitutes the whole match when the pattern has no capture group', async () => { + const streamed = await streamScripted('{"goal_id":"{{fromRequest:goal-[0-9a-z]+}}"}') + const delta = streamed.find(chunk => chunk.type === 'tool-call-delta') + expect(delta).toMatchObject({ argumentsDelta: '{"goal_id":"goal-42ab"}' }) + }) + + it('fails loud when a placeholder matches nothing in the request', async () => { + await expect(streamScripted('{"goal_id":"{{fromRequest:task-[0-9]+}}"}')) + .rejects.toThrow(/fromRequest.*matched nothing/) + }) + + it('fails loud on an invalid placeholder pattern', async () => { + await expect(streamScripted('{"goal_id":"{{fromRequest:(goal-}}"}')) + .rejects.toThrow(/fromRequest.*invalid pattern/) + }) + + it('fails loud on an unterminated placeholder', () => { + const entry: ReplayEntry = { kind: 'chunks', chunks: scriptedCall('{"goal_id":"{{fromRequest:goal-1"}') } + expect(() => resolveScriptedEntry(entry, requestMessages)).toThrow(/fromRequest placeholder is unterminated/) + }) + + it('returns the exact same entry when no placeholder appears', () => { + const entry: ReplayEntry = { kind: 'chunks', chunks: TEXT_CHUNKS } + expect(resolveScriptedEntry(entry, requestMessages)).toBe(entry) + }) + + it('skips non-string request leaves when building the corpus', () => { + const messages = requestMessages.map(message => ({ ...message, seq: 7 })) as unknown as GenerateOptions['messages'] + const entry: ReplayEntry = { kind: 'chunks', chunks: scriptedCall('{"goal_id":"{{fromRequest:goal-42[a-z]+}}"}') } + const resolved = resolveScriptedEntry(entry, messages) + if (resolved.kind !== 'chunks') throw new Error('expected chunks entry') + expect(resolved.chunks[1]).toMatchObject({ argumentsDelta: '{"goal_id":"goal-42ab"}' }) + }) + }) + it('registers a replay-only provider catalog when configured', async () => { writeLog(TEXT_CHUNKS) const ctx = new Context()