diff --git a/apps/server/src/ru-fork/qwen-transcript/QwenTranscriptLive.ts b/apps/server/src/ru-fork/qwen-transcript/QwenTranscriptLive.ts new file mode 100644 index 00000000000..0f286bc4bbe --- /dev/null +++ b/apps/server/src/ru-fork/qwen-transcript/QwenTranscriptLive.ts @@ -0,0 +1,139 @@ +// ru-fork: Live layer for QwenTranscriptService. Resolves a thread's transcript +// path from durable state (cwd via ProjectionSnapshotQuery.getThreadCheckpointContext, +// sessionId via ProviderSessionDirectory.getBinding), then tails the append-only +// JSONL with the proven serverSettings.ts watch+debounce pattern. Race-free and +// dedup-free: the watch is attached before the first read and every emission is a +// slice from one monotonic counter. See ru-fork-instrumental/advanced-chat/PLAN.md. +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; + +import type { ThreadId, TranscriptStreamItem } from "@t3tools/contracts"; +import { TranscriptSubscribeError } from "@t3tools/contracts"; + +import { ServerConfig } from "../../config.ts"; +import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts"; +import { ProviderSessionDirectory } from "../../provider/Services/ProviderSessionDirectory.ts"; +import { readRuntimeOutputDirOverride } from "./cliSettings.ts"; +import { parseAndNormalize } from "./parse.ts"; +import { resolveChatsDir } from "./paths.ts"; +import { QwenTranscriptService, type QwenTranscriptServiceShape } from "./QwenTranscriptService.ts"; + +/** Parse our `CliResumeCursor` ({ schemaVersion: 1, sessionId }) → sessionId. */ +const parseResumeSessionId = (raw: unknown): Option.Option => { + if (typeof raw !== "object" || raw === null) return Option.none(); + const candidate = raw as { schemaVersion?: unknown; sessionId?: unknown }; + if (candidate.schemaVersion !== 1) return Option.none(); + return typeof candidate.sessionId === "string" && candidate.sessionId.length > 0 + ? Option.some(candidate.sessionId) + : Option.none(); +}; + +export const QwenTranscriptLive = Layer.effect( + QwenTranscriptService, + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const config = yield* ServerConfig; + const directory = yield* ProviderSessionDirectory; + const projection = yield* ProjectionSnapshotQuery; + + const resolveSessionId = (threadId: ThreadId): Effect.Effect> => + directory.getBinding(threadId).pipe( + Effect.map((binding) => + Option.isSome(binding) ? parseResumeSessionId(binding.value.resumeCursor) : Option.none(), + ), + Effect.orElseSucceed(() => Option.none()), + ); + + const subscribe = ( + threadId: ThreadId, + ): Stream.Stream => + Stream.unwrap( + Effect.gen(function* () { + const ctx = yield* projection.getThreadCheckpointContext(threadId).pipe( + Effect.mapError( + (cause) => + new TranscriptSubscribeError({ + message: `Не удалось определить рабочий каталог потока: ${String(cause)}`, + threadId, + }), + ), + ); + if (Option.isNone(ctx)) { + return yield* new TranscriptSubscribeError({ + message: `Поток ${threadId} не найден.`, + threadId, + }); + } + const cwd = ctx.value.worktreePath ?? ctx.value.workspaceRoot; + const runtimeOutputDirSetting = yield* readRuntimeOutputDirOverride( + fs, + path, + config.cliConfigDir, + cwd, + ); + const baseInput = { + env: process.env, + cliConfigDir: config.cliConfigDir, + cwd, + runtimeOutputDirSetting, + }; + const chatsDir = resolveChatsDir(baseInput); + yield* fs.makeDirectory(chatsDir, { recursive: true }).pipe(Effect.ignore); + + const emitted = yield* Ref.make(0); + + // Read + normalize all records currently on disk for a known session. + // Missing session id or missing file → no records (an empty feed). + const readRecords = (sessionId: Option.Option) => + Effect.gen(function* () { + if (Option.isNone(sessionId)) return []; + const filePath = path.join(chatsDir, `${sessionId.value}.jsonl`); + const exists = yield* fs.exists(filePath).pipe(Effect.orElseSucceed(() => false)); + const text = exists + ? yield* fs.readFileString(filePath).pipe(Effect.orElseSucceed(() => "")) + : ""; + return parseAndNormalize(text); + }); + + // First non-empty read is a `snapshot`; every later read is an `append` + // of only the records beyond what we've already emitted. + const readDelta: Effect.Effect = Effect.gen(function* () { + const sessionId = yield* resolveSessionId(threadId); + const records = yield* readRecords(sessionId); + const prev = yield* Ref.getAndSet(emitted, records.length); + return prev === 0 + ? { kind: "snapshot" as const, records } + : { kind: "append" as const, records: records.slice(prev) }; + }); + + // Hybrid liveness: fs.watch gives instant updates where inotify works; + // a low-frequency poll guarantees eventual delivery on filesystems that + // don't deliver watch events (network/virtual mounts, some containers). + // Each trigger re-reads + slices, so duplicate triggers are harmless and + // any append missed in the watch-attach window is healed by the next poll. + const watchTriggers = fs.watch(chatsDir).pipe( + Stream.filter((event) => event.path.endsWith(".jsonl")), + Stream.debounce(Duration.millis(100)), + Stream.catchCause(() => Stream.empty), + ); + const pollTriggers = Stream.tick(Duration.seconds(1)); + const liveDeltas = Stream.merge(watchTriggers, pollTriggers).pipe( + Stream.mapEffect(() => readDelta), + ); + + return Stream.concat(Stream.fromEffect(readDelta), liveDeltas).pipe( + Stream.filter((item) => item.kind === "snapshot" || item.records.length > 0), + ); + }), + ); + + return { subscribe } satisfies QwenTranscriptServiceShape; + }), +); diff --git a/apps/server/src/ru-fork/qwen-transcript/QwenTranscriptService.ts b/apps/server/src/ru-fork/qwen-transcript/QwenTranscriptService.ts new file mode 100644 index 00000000000..3d79883c2ee --- /dev/null +++ b/apps/server/src/ru-fork/qwen-transcript/QwenTranscriptService.ts @@ -0,0 +1,23 @@ +// ru-fork: advanced chat mode — service that streams a thread's normalized qwen +// transcript (snapshot + live appends). Read-only; the control plane (approvals, +// turns) stays on the ACP path. See ru-fork-instrumental/advanced-chat/PLAN.md. +import * as Context from "effect/Context"; +import type * as Stream from "effect/Stream"; + +import type { ThreadId, TranscriptStreamItem, TranscriptSubscribeError } from "@t3tools/contracts"; + +export interface QwenTranscriptServiceShape { + /** + * Stream the transcript for `threadId`: first item is a `snapshot` of all + * records currently on disk; subsequent items are `append` batches as the CLI + * writes. Fails only when the thread itself cannot be resolved. + */ + readonly subscribe: ( + threadId: ThreadId, + ) => Stream.Stream; +} + +export class QwenTranscriptService extends Context.Service< + QwenTranscriptService, + QwenTranscriptServiceShape +>()("t3/ru-fork/QwenTranscriptService") {} diff --git a/apps/server/src/ru-fork/qwen-transcript/cliSettings.ts b/apps/server/src/ru-fork/qwen-transcript/cliSettings.ts new file mode 100644 index 00000000000..215a490939a --- /dev/null +++ b/apps/server/src/ru-fork/qwen-transcript/cliSettings.ts @@ -0,0 +1,55 @@ +// ru-fork: read `advanced.runtimeOutputDir` from the spawned CLI's settings.json +// (global `/settings.json`, then workspace `//settings.json`, +// workspace wins — qwen's merge order). Missing/invalid → undefined. This is the +// ONLY thing that moves transcripts off the default base (see paths.ts). +// +// Takes already-resolved `fs`/`path` VALUES (not service requirements) so the +// returned Effect has R = never and the transcript stream stays requirement-free. +// +// We parse the CLI's own settings.json (a foreign file with an unknown shape we +// only sample one field from), so a defensive JSON.parse is the right tool here +// rather than a full Effect Schema decode. +// @effect-diagnostics preferSchemaOverJson:off +import * as Effect from "effect/Effect"; +import type * as FileSystem from "effect/FileSystem"; +import type * as Path from "effect/Path"; + +import { CLI_FOLDER } from "../../config.ts"; + +/** Pure: extract `advanced.runtimeOutputDir` from a parsed settings object. */ +export const pickRuntimeOutputDir = (raw: unknown): string | undefined => { + if (typeof raw !== "object" || raw === null) return undefined; + const advanced = (raw as Record)["advanced"]; + if (typeof advanced !== "object" || advanced === null) return undefined; + const value = (advanced as Record)["runtimeOutputDir"]; + return typeof value === "string" && value.trim().length > 0 ? value : undefined; +}; + +/** Pure JSON parse that never throws (malformed → undefined). */ +const parseJsonSafe = (text: string): unknown => { + try { + return JSON.parse(text) as unknown; + } catch { + return undefined; + } +}; + +const readJsonSafe = (fs: FileSystem.FileSystem, filePath: string): Effect.Effect => + Effect.gen(function* () { + const exists = yield* fs.exists(filePath).pipe(Effect.orElseSucceed(() => false)); + if (!exists) return undefined; + const text = yield* fs.readFileString(filePath).pipe(Effect.orElseSucceed(() => "")); + return parseJsonSafe(text); + }); + +export const readRuntimeOutputDirOverride = ( + fs: FileSystem.FileSystem, + path: Path.Path, + cliConfigDir: string, + cwd: string, +): Effect.Effect => + Effect.gen(function* () { + const globalRaw = yield* readJsonSafe(fs, path.join(cliConfigDir, "settings.json")); + const workspaceRaw = yield* readJsonSafe(fs, path.join(cwd, CLI_FOLDER, "settings.json")); + return pickRuntimeOutputDir(workspaceRaw) ?? pickRuntimeOutputDir(globalRaw); + }); diff --git a/apps/server/src/ru-fork/qwen-transcript/parse.ts b/apps/server/src/ru-fork/qwen-transcript/parse.ts new file mode 100644 index 00000000000..1ed8a5ead71 --- /dev/null +++ b/apps/server/src/ru-fork/qwen-transcript/parse.ts @@ -0,0 +1,259 @@ +// ru-fork: pure parse + normalize of qwen's JSONL transcript records into the +// `TranscriptRecord` wire contract. No I/O, no throws — malformed input is +// skipped/counted so one bad line can never poison the feed. Mirrors qwen +// chatRecordingService.ts (ChatRecord), tools.ts (ToolResultDisplay), and +// geminiChat.ts (Part discrimination). +import type { + TranscriptPart, + TranscriptRecord, + TranscriptToolCall, + TranscriptToolDisplay, + TranscriptToolStatus, + TranscriptUsage, +} from "@t3tools/contracts"; + +type Json = unknown; +type Obj = Record; + +const isObj = (value: unknown): value is Obj => + typeof value === "object" && value !== null && !Array.isArray(value); + +const asString = (value: unknown): string | undefined => + typeof value === "string" ? value : undefined; + +const asNumber = (value: unknown): number | undefined => + typeof value === "number" && Number.isFinite(value) ? value : undefined; + +export interface ParseResult { + readonly records: TranscriptRecord[]; + readonly errorCount: number; +} + +/** Split append-only JSONL → parsed objects. Trailing partial/blank lines and + * malformed lines are skipped (the latter counted). */ +export const parseTranscriptJsonl = (text: string): { rows: Obj[]; errorCount: number } => { + const rows: Obj[] = []; + let errorCount = 0; + for (const line of text.split("\n")) { + const trimmed = line.trim(); + if (trimmed.length === 0) continue; + try { + const parsed = JSON.parse(trimmed) as unknown; + if (isObj(parsed)) rows.push(parsed); + else errorCount += 1; + } catch { + errorCount += 1; + } + } + return { rows, errorCount }; +}; + +const normalizePart = (raw: unknown): TranscriptPart => { + if (!isObj(raw)) return { kind: "unknown", raw }; + if (raw["thought"] === true) { + return { kind: "thought", text: asString(raw["text"]) ?? "" }; + } + if (isObj(raw["functionCall"])) { + const fc = raw["functionCall"]; + return { kind: "function_call", name: asString(fc["name"]) ?? "", args: fc["args"] }; + } + if (isObj(raw["functionResponse"])) { + const fr = raw["functionResponse"]; + return { + kind: "function_response", + name: asString(fr["name"]) ?? "", + response: fr["response"], + }; + } + if (typeof raw["text"] === "string") { + return { kind: "text", text: raw["text"] }; + } + if (isObj(raw["inlineData"])) { + return { kind: "inline_data", mimeType: asString(raw["inlineData"]["mimeType"]) ?? "" }; + } + return { kind: "unknown", raw }; +}; + +const normalizeParts = (message: unknown): TranscriptPart[] => { + if (!isObj(message)) return []; + const parts = message["parts"]; + if (!Array.isArray(parts)) return []; + return parts.map(normalizePart); +}; + +const normalizeUsage = (raw: unknown): TranscriptUsage | undefined => { + if (!isObj(raw)) return undefined; + const prompt = asNumber(raw["promptTokenCount"]); + const cached = asNumber(raw["cachedInputTokenCount"]); + const output = asNumber(raw["candidatesTokenCount"]); + const total = asNumber(raw["totalTokenCount"]); + const usage: TranscriptUsage = { + ...(prompt !== undefined ? { promptTokens: prompt } : {}), + ...(cached !== undefined ? { cachedTokens: cached } : {}), + ...(output !== undefined ? { outputTokens: output } : {}), + ...(total !== undefined ? { totalTokens: total } : {}), + }; + return Object.keys(usage).length > 0 ? usage : undefined; +}; + +const TOOL_STATUSES: ReadonlySet = new Set([ + "validating", + "scheduled", + "error", + "success", + "executing", + "cancelled", + "awaiting_approval", +]); + +const normalizeDisplay = (raw: unknown): TranscriptToolDisplay | undefined => { + if (raw === undefined || raw === null) return undefined; + if (typeof raw === "string") return { kind: "text", text: raw }; + if (!isObj(raw)) return { kind: "unknown", raw }; + + if (typeof raw["fileDiff"] === "string") { + return { + kind: "file_diff", + fileName: asString(raw["fileName"]) ?? "", + fileDiff: raw["fileDiff"], + originalContent: typeof raw["originalContent"] === "string" ? raw["originalContent"] : null, + newContent: asString(raw["newContent"]) ?? "", + ...(raw["diffStat"] !== undefined ? { diffStat: raw["diffStat"] } : {}), + }; + } + + const type = raw["type"]; + if (type === "todo_list" && Array.isArray(raw["todos"])) { + const todos = raw["todos"] + .filter(isObj) + .map((todo) => ({ + id: asString(todo["id"]) ?? "", + content: asString(todo["content"]) ?? "", + status: ((): "pending" | "in_progress" | "completed" => { + const status = todo["status"]; + return status === "in_progress" || status === "completed" ? status : "pending"; + })(), + })); + return { kind: "todo_list", todos }; + } + if (type === "plan_summary") { + return { + kind: "plan_summary", + message: asString(raw["message"]) ?? "", + plan: asString(raw["plan"]) ?? "", + ...(typeof raw["rejected"] === "boolean" ? { rejected: raw["rejected"] } : {}), + }; + } + if (type === "task_execution") { + return { + kind: "task_execution", + subagentName: asString(raw["subagentName"]) ?? "", + status: asString(raw["status"]) ?? "", + taskDescription: asString(raw["taskDescription"]) ?? "", + ...(asString(raw["result"]) !== undefined ? { result: asString(raw["result"]) } : {}), + ...(Array.isArray(raw["toolCalls"]) ? { toolCalls: raw["toolCalls"] } : {}), + }; + } + if (type === "mcp_tool_progress") { + return { + kind: "mcp_progress", + progress: asNumber(raw["progress"]) ?? 0, + ...(asNumber(raw["total"]) !== undefined ? { total: asNumber(raw["total"]) } : {}), + ...(asString(raw["message"]) !== undefined ? { message: asString(raw["message"]) } : {}), + }; + } + if (raw["ansiOutput"] !== undefined) { + return { kind: "ansi", raw: raw["ansiOutput"] }; + } + return { kind: "unknown", raw }; +}; + +const normalizeToolCall = (raw: unknown): TranscriptToolCall | undefined => { + if (!isObj(raw)) return undefined; + const callId = asString(raw["callId"]); + const statusRaw = raw["status"]; + const status = + typeof statusRaw === "string" && TOOL_STATUSES.has(statusRaw) + ? (statusRaw as TranscriptToolStatus) + : undefined; + const display = normalizeDisplay(raw["resultDisplay"]); + return { + ...(callId !== undefined ? { callId } : {}), + ...(status !== undefined ? { status } : {}), + ...(display !== undefined ? { display } : {}), + }; +}; + +const SYSTEM_SUBTYPES: ReadonlySet = new Set([ + "chat_compression", + "slash_command", + "ui_telemetry", + "at_command", +]); + +/** qwen ChatRecord → TranscriptRecord. Returns null for an unusable row. */ +export const normalizeRecord = (raw: Obj): TranscriptRecord | null => { + const type = raw["type"]; + const uuid = asString(raw["uuid"]); + const sessionId = asString(raw["sessionId"]); + const timestamp = asString(raw["timestamp"]); + if (uuid === undefined || sessionId === undefined || timestamp === undefined) return null; + + const base = { + uuid, + parentUuid: typeof raw["parentUuid"] === "string" ? raw["parentUuid"] : null, + sessionId, + timestamp, + cwd: asString(raw["cwd"]) ?? "", + ...(asString(raw["gitBranch"]) !== undefined ? { gitBranch: asString(raw["gitBranch"]) } : {}), + }; + + if (type === "user") { + return { ...base, type: "user", parts: normalizeParts(raw["message"]) }; + } + if (type === "assistant") { + const usage = normalizeUsage(raw["usageMetadata"]); + return { + ...base, + type: "assistant", + parts: normalizeParts(raw["message"]), + ...(asString(raw["model"]) !== undefined ? { model: asString(raw["model"]) } : {}), + ...(asNumber(raw["contextWindowSize"]) !== undefined + ? { contextWindowSize: asNumber(raw["contextWindowSize"]) } + : {}), + ...(usage !== undefined ? { usage } : {}), + }; + } + if (type === "tool_result") { + const toolCall = normalizeToolCall(raw["toolCallResult"]); + return { + ...base, + type: "tool_result", + parts: normalizeParts(raw["message"]), + ...(toolCall !== undefined ? { toolCall } : {}), + }; + } + if (type === "system") { + const subtype = raw["subtype"]; + return { + ...base, + type: "system", + ...(typeof subtype === "string" && SYSTEM_SUBTYPES.has(subtype) + ? { subtype: subtype as "chat_compression" | "slash_command" | "ui_telemetry" | "at_command" } + : {}), + ...(raw["systemPayload"] !== undefined ? { payload: raw["systemPayload"] as Json } : {}), + }; + } + return null; +}; + +/** Full pipeline: JSONL text → ordered normalized records. */ +export const parseAndNormalize = (text: string): TranscriptRecord[] => { + const { rows } = parseTranscriptJsonl(text); + const records: TranscriptRecord[] = []; + for (const row of rows) { + const normalized = normalizeRecord(row); + if (normalized !== null) records.push(normalized); + } + return records; +}; diff --git a/apps/server/src/ru-fork/qwen-transcript/paths.ts b/apps/server/src/ru-fork/qwen-transcript/paths.ts new file mode 100644 index 00000000000..d81333efe90 --- /dev/null +++ b/apps/server/src/ru-fork/qwen-transcript/paths.ts @@ -0,0 +1,65 @@ +// @effect-diagnostics nodeBuiltinImport:off +// ru-fork: deterministic resolver for qwen's on-disk transcript file. Mirrors +// qwen-code core/src/config/storage.ts (getRuntimeBaseDir / getProjectDir) and +// utils/paths.ts (sanitizeCwd) EXACTLY so our resolved path === the CLI's path. +// +// Base dir priority (storage.ts:102-118): +// QWEN_RUNTIME_DIR env > settings.advanced.runtimeOutputDir > getGlobalQwenDir() +// where getGlobalQwenDir() == os.homedir()/$CLI_DIR == ServerConfig.cliConfigDir. +// We use `cliConfigDir` for the default so the base tracks our resolved CLI +// config dir (the same root the skills scanner reads), never a literal ".qwen". +import * as path from "node:path"; +import * as os from "node:os"; + +/** 1:1 port of qwen utils/paths.ts:218 sanitizeCwd. `platform` is injectable for tests. */ +export const sanitizeCwd = (cwd: string, platform: NodeJS.Platform = os.platform()): string => { + const normalized = platform === "win32" ? cwd.toLowerCase() : cwd; + return normalized.replace(/[^a-zA-Z0-9]/g, "-"); +}; + +/** 1:1 port of qwen storage.ts:40-67 resolveRuntimeBaseDir (tilde + relative). */ +export const expandRuntimeBaseDir = (dir: string, cwd: string): string => { + let resolved = dir; + if (resolved === "~" || resolved.startsWith("~/") || resolved.startsWith("~\\")) { + const rest = + resolved === "~" ? [] : resolved.slice(2).split(/[/\\]+/).filter(Boolean); + resolved = path.join(os.homedir(), ...rest); + } + if (!path.isAbsolute(resolved)) { + resolved = path.resolve(cwd, resolved); + } + return resolved; +}; + +export interface TranscriptBaseInput { + /** process env of THIS server (the child inherits it; QWEN_RUNTIME_DIR wins). */ + readonly env: NodeJS.ProcessEnv; + /** ServerConfig.cliConfigDir = {home}/$CLI_DIR = qwen's default runtime base. */ + readonly cliConfigDir: string; + /** thread cwd (worktreePath ?? workspaceRoot). */ + readonly cwd: string; + /** advanced.runtimeOutputDir from the CLI settings.json, if any. */ + readonly runtimeOutputDirSetting: string | undefined; +} + +/** Mirror of storage.ts:102-118 priority. */ +export const resolveTranscriptBaseDir = (input: TranscriptBaseInput): string => { + const fromEnv = input.env["QWEN_RUNTIME_DIR"]; + if (fromEnv && fromEnv.trim()) { + return expandRuntimeBaseDir(fromEnv.trim(), input.cwd); + } + const setting = input.runtimeOutputDirSetting?.trim(); + if (setting) { + return expandRuntimeBaseDir(setting, input.cwd); + } + return input.cliConfigDir; +}; + +/** `/projects//chats`. */ +export const resolveChatsDir = (input: TranscriptBaseInput): string => + path.join(resolveTranscriptBaseDir(input), "projects", sanitizeCwd(input.cwd), "chats"); + +/** `/.jsonl`. */ +export const resolveTranscriptFilePath = ( + input: TranscriptBaseInput & { readonly sessionId: string }, +): string => path.join(resolveChatsDir(input), `${input.sessionId}.jsonl`); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 707fec94e9b..51951dbe06b 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -30,6 +30,7 @@ import { FilesystemBrowseError, ThreadId, type TerminalEvent, + TRANSCRIPT_WS_METHOD, WS_METHODS, WsRpcGroup, } from "@t3tools/contracts"; @@ -49,6 +50,9 @@ import { McpProjectionQuery } from "./ru-fork/mcp/McpProjectionQuery.ts"; import { McpRuntime } from "./ru-fork/mcp/McpRuntime.ts"; import { McpSupervisor } from "./ru-fork/mcp/McpSupervisor.ts"; import { ProjectionSnapshotQuery } from "./orchestration/Services/ProjectionSnapshotQuery.ts"; +// ru-fork: advanced chat mode — qwen transcript stream service. +import { QwenTranscriptService } from "./ru-fork/qwen-transcript/QwenTranscriptService.ts"; +import { QwenTranscriptLive } from "./ru-fork/qwen-transcript/QwenTranscriptLive.ts"; // ru-fork: telemetry/observability (spans + metrics) was removed, and the // per-RPC FAILURE CAPTURE that lived in this wrapper went with it — so every // RPC failed silently server-side. These wrappers restore just the failure @@ -194,6 +198,8 @@ const makeWsRpcLayer = (currentSessionId: AuthSessionId) => WsRpcGroup.toLayer( Effect.gen(function* () { const projectionSnapshotQuery = yield* ProjectionSnapshotQuery; + // ru-fork: advanced chat mode — read-only qwen transcript feed. + const qwenTranscript = yield* QwenTranscriptService; const orchestrationEngine = yield* OrchestrationEngineService; const checkpointDiffQuery = yield* CheckpointDiffQuery; const keybindings = yield* Keybindings; @@ -886,6 +892,13 @@ const makeWsRpcLayer = (currentSessionId: AuthSessionId) => }), { "rpc.aggregate": "orchestration" }, ), + // ru-fork: advanced chat mode — stream the qwen JSONL transcript. + [TRANSCRIPT_WS_METHOD]: (input) => + observeRpcStream( + TRANSCRIPT_WS_METHOD, + qwenTranscript.subscribe(input.threadId), + { "rpc.aggregate": "orchestration" }, + ), [WS_METHODS.serverGetConfig]: (_input) => observeRpcEffect(WS_METHODS.serverGetConfig, loadServerConfig, { "rpc.aggregate": "server", @@ -1345,6 +1358,10 @@ export const websocketRpcRouteLayer = prefixedRouteLayer( Effect.provide( makeWsRpcLayer(session.sessionId).pipe( Layer.provideMerge(RpcSerialization.layerJson), + // ru-fork: advanced chat mode — deps (ProjectionSnapshotQuery, + // ProviderSessionDirectory, ServerConfig, FileSystem) come from the + // outer runtime context this route is provided within. + Layer.provide(QwenTranscriptLive), Layer.provide(ProviderMaintenanceRunner.layer), Layer.provide( SourceControlDiscoveryLayer.layer.pipe( diff --git a/apps/server/tests/ru-fork/qwen-transcript/QwenTranscriptLive.test.ts b/apps/server/tests/ru-fork/qwen-transcript/QwenTranscriptLive.test.ts new file mode 100644 index 00000000000..a43e94fdf35 --- /dev/null +++ b/apps/server/tests/ru-fork/qwen-transcript/QwenTranscriptLive.test.ts @@ -0,0 +1,338 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { expect, it } from "@effect/vitest"; +import { + ProjectId, + ProviderDriverKind, + ThreadId, + type TranscriptStreamItem, +} from "@t3tools/contracts"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Stream from "effect/Stream"; + +import { ServerConfig } from "../../../src/config.ts"; +import { + ProjectionSnapshotQuery, + type ProjectionSnapshotQueryShape, +} from "../../../src/orchestration/Services/ProjectionSnapshotQuery.ts"; +import { + ProviderSessionDirectory, + type ProviderSessionDirectoryShape, +} from "../../../src/provider/Services/ProviderSessionDirectory.ts"; +import { QwenTranscriptLive } from "../../../src/ru-fork/qwen-transcript/QwenTranscriptLive.ts"; +import { QwenTranscriptService } from "../../../src/ru-fork/qwen-transcript/QwenTranscriptService.ts"; +import { resolveTranscriptFilePath } from "../../../src/ru-fork/qwen-transcript/paths.ts"; + +const THREAD_ID = ThreadId.make("thread-x"); + +const userLine = (sessionId: string, cwd: string, uuid = "u1") => + JSON.stringify({ + uuid, + parentUuid: null, + sessionId, + timestamp: "2026-01-01T00:00:00.000Z", + cwd, + type: "user", + message: { parts: [{ text: "hi" }] }, + }); + +const assistantLine = (sessionId: string, cwd: string, uuid = "a1") => + JSON.stringify({ + uuid, + parentUuid: "u1", + sessionId, + timestamp: "2026-01-01T00:00:01.000Z", + cwd, + type: "assistant", + model: "claude-opus-4", + message: { parts: [{ text: "hello" }] }, + }); + +// Mock projection: optionally resolve a thread cwd (workspaceRoot + worktreePath). +const projectionLayer = ( + ctx: Option.Option | "fail", + worktreePath: string | null = null, +) => + Layer.succeed(ProjectionSnapshotQuery, { + getThreadCheckpointContext: () => + ctx === "fail" + ? Effect.fail("boom") + : Effect.succeed( + Option.map(ctx, (workspaceRoot) => ({ + threadId: THREAD_ID, + projectId: ProjectId.make("p1"), + workspaceRoot, + worktreePath, + checkpoints: [], + })), + ), + } as unknown as ProjectionSnapshotQueryShape); + +// Mock directory: optionally expose a resume sessionId. +const directoryLayer = (sessionId: string | null) => + directoryWith(sessionId ? { schemaVersion: 1, sessionId } : null); + +// Mock directory with an arbitrary resume cursor, or no binding at all. +const directoryWith = (resumeCursor: unknown | "no-binding") => + Layer.succeed(ProviderSessionDirectory, { + getBinding: () => + Effect.succeed( + resumeCursor === "no-binding" + ? Option.none() + : Option.some({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("cli"), + resumeCursor, + }), + ), + } as unknown as ProviderSessionDirectoryShape); + +const TestLayer = ServerConfig.layerTest(process.cwd(), { prefix: "t3-transcript-cfg-" }).pipe( + Layer.provideMerge(NodeServices.layer), +); + +it.layer(TestLayer)("QwenTranscriptLive", (it) => { + // Build a fresh service with the given mocked thread cwd + sessionId. + const getService = (cwd: Option.Option | "fail", sessionId: string | null) => + Effect.service(QwenTranscriptService).pipe( + Effect.provide( + QwenTranscriptLive.pipe( + Layer.provide(projectionLayer(cwd)), + Layer.provide(directoryLayer(sessionId)), + ), + ), + ); + + const filePathFor = (cwd: string, sessionId: string, runtimeOutputDirSetting?: string) => + Effect.gen(function* () { + const config = yield* ServerConfig; + return resolveTranscriptFilePath({ + env: process.env, + cliConfigDir: config.cliConfigDir, + cwd, + runtimeOutputDirSetting, + sessionId, + }); + }); + + const writeFile = (filePath: string, content: string) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + yield* fs.makeDirectory(path.dirname(filePath), { recursive: true }); + yield* fs.writeFileString(filePath, content); + }); + + const mkCwd = Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + return yield* fs.makeTempDirectoryScoped({ prefix: "t3-transcript-cwd-" }); + }); + + it.effect("emits a snapshot of an existing transcript", () => + Effect.gen(function* () { + const cwd = yield* mkCwd; + const sessionId = "sess-1"; + const filePath = yield* filePathFor(cwd, sessionId); + yield* writeFile(filePath, `${userLine(sessionId, cwd)}\n${assistantLine(sessionId, cwd)}\n`); + + const svc = yield* getService(Option.some(cwd), sessionId); + const head = yield* Stream.runHead(svc.subscribe(THREAD_ID)); + + const item = Option.getOrThrow(head); + expect(item.kind).toBe("snapshot"); + expect(item.records.map((r) => r.type)).toEqual(["user", "assistant"]); + expect(item.records[0]).toMatchObject({ parts: [{ kind: "text", text: "hi" }] }); + }), + ); + + it.effect("emits an empty snapshot when the file does not exist yet", () => + Effect.gen(function* () { + const cwd = yield* mkCwd; + const svc = yield* getService(Option.some(cwd), "sess-missing"); + const head = yield* Stream.runHead(svc.subscribe(THREAD_ID)); + expect(Option.getOrThrow(head)).toEqual({ kind: "snapshot", records: [] }); + }), + ); + + it.effect("emits an empty snapshot when the session id is not known yet", () => + Effect.gen(function* () { + const cwd = yield* mkCwd; + const svc = yield* getService(Option.some(cwd), null); + const head = yield* Stream.runHead(svc.subscribe(THREAD_ID)); + expect(Option.getOrThrow(head)).toEqual({ kind: "snapshot", records: [] }); + }), + ); + + it.effect("skips malformed lines and stays alive", () => + Effect.gen(function* () { + const cwd = yield* mkCwd; + const sessionId = "sess-bad"; + const filePath = yield* filePathFor(cwd, sessionId); + yield* writeFile(filePath, `${userLine(sessionId, cwd)}\n{not json}\n`); + + const svc = yield* getService(Option.some(cwd), sessionId); + const head = yield* Stream.runHead(svc.subscribe(THREAD_ID)); + const item = Option.getOrThrow(head); + expect(item.kind).toBe("snapshot"); + expect(item.records).toHaveLength(1); + }), + ); + + it.effect("fails the stream when the thread cannot be resolved", () => + Effect.gen(function* () { + const svc = yield* getService(Option.none(), "sess-x"); + const exit = yield* Effect.exit(Stream.runHead(svc.subscribe(THREAD_ID))); + expect(Exit.isFailure(exit)).toBe(true); + }), + ); + + it.effect("fails the stream when the cwd lookup errors", () => + Effect.gen(function* () { + const svc = yield* getService("fail", "sess-x"); + const exit = yield* Effect.exit(Stream.runHead(svc.subscribe(THREAD_ID))); + expect(Exit.isFailure(exit)).toBe(true); + }), + ); + + it.effect("reads from advanced.runtimeOutputDir override, not cliConfigDir", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const cwd = yield* mkCwd; + const overrideDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-transcript-rt-" }); + const sessionId = "sess-override"; + + // Workspace settings.json points the runtime base at overrideDir. + yield* fs.makeDirectory(path.join(cwd, ".qwen"), { recursive: true }); + yield* fs.writeFileString( + path.join(cwd, ".qwen", "settings.json"), + JSON.stringify({ advanced: { runtimeOutputDir: overrideDir } }), + ); + + // Transcript written under the OVERRIDE dir. + const filePath = yield* filePathFor(cwd, sessionId, overrideDir); + yield* writeFile(filePath, `${userLine(sessionId, cwd)}\n`); + + const svc = yield* getService(Option.some(cwd), sessionId); + const head = yield* Stream.runHead(svc.subscribe(THREAD_ID)); + const item = Option.getOrThrow(head); + expect(item.kind).toBe("snapshot"); + expect(item.records).toHaveLength(1); + }), + ); + + it.effect("uses worktreePath when present (over workspaceRoot)", () => + Effect.gen(function* () { + const worktree = yield* mkCwd; + const sessionId = "sess-wt"; + const filePath = yield* filePathFor(worktree, sessionId); + yield* writeFile(filePath, `${userLine(sessionId, worktree)}\n`); + + const svc = yield* Effect.service(QwenTranscriptService).pipe( + Effect.provide( + QwenTranscriptLive.pipe( + // workspaceRoot is a bogus dir; worktreePath is where the file lives. + Layer.provide(projectionLayer(Option.some("/nonexistent-workspace"), worktree)), + Layer.provide(directoryLayer(sessionId)), + ), + ), + ); + const head = yield* Stream.runHead(svc.subscribe(THREAD_ID)); + expect(Option.getOrThrow(head).records).toHaveLength(1); + }), + ); + + it.effect("emits an empty snapshot when no provider binding exists", () => + Effect.gen(function* () { + const cwd = yield* mkCwd; + const svc = yield* Effect.service(QwenTranscriptService).pipe( + Effect.provide( + QwenTranscriptLive.pipe( + Layer.provide(projectionLayer(Option.some(cwd))), + Layer.provide(directoryWith("no-binding")), + ), + ), + ); + const head = yield* Stream.runHead(svc.subscribe(THREAD_ID)); + expect(Option.getOrThrow(head)).toEqual({ kind: "snapshot", records: [] }); + }), + ); + + it.effect("treats unparseable resume cursors as no session", () => + Effect.gen(function* () { + const cwd = yield* mkCwd; + const badCursors: unknown[] = [ + { schemaVersion: 2, sessionId: "x" }, + { schemaVersion: 1, sessionId: 123 }, + { schemaVersion: 1, sessionId: "" }, + ]; + for (const resumeCursor of badCursors) { + const svc = yield* Effect.service(QwenTranscriptService).pipe( + Effect.provide( + QwenTranscriptLive.pipe( + Layer.provide(projectionLayer(Option.some(cwd))), + Layer.provide(directoryWith(resumeCursor)), + ), + ), + ); + const head = yield* Stream.runHead(svc.subscribe(THREAD_ID)); + expect(Option.getOrThrow(head)).toEqual({ kind: "snapshot", records: [] }); + } + }), + ); +}); + +// Real-clock test: fs.watch / poll / sleep advance in wall-clock time, so we +// observe a genuine live append (it.live runs outside the it.layer context, so +// this test provides its own layer + Scope). +it.live("streams a live append after the snapshot", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const config = yield* ServerConfig; + const cwd = yield* fs.makeTempDirectoryScoped({ prefix: "t3-transcript-live-" }); + const sessionId = "sess-live"; + const filePath = resolveTranscriptFilePath({ + env: process.env, + cliConfigDir: config.cliConfigDir, + cwd, + runtimeOutputDirSetting: undefined, + sessionId, + }); + yield* fs.makeDirectory(path.dirname(filePath), { recursive: true }); + yield* fs.writeFileString(filePath, `${userLine(sessionId, cwd)}\n`); + + const svc = yield* Effect.service(QwenTranscriptService).pipe( + Effect.provide( + QwenTranscriptLive.pipe( + Layer.provide(projectionLayer(Option.some(cwd))), + Layer.provide(directoryLayer(sessionId)), + ), + ), + ); + + // Append a record shortly after subscribing; the collector blocks until 2 items. + yield* Effect.forkScoped( + Effect.gen(function* () { + yield* Effect.sleep(Duration.millis(300)); + yield* fs.writeFileString( + filePath, + `${userLine(sessionId, cwd)}\n${assistantLine(sessionId, cwd)}\n`, + ); + }), + ); + + const items = (yield* Stream.runCollect( + Stream.take(svc.subscribe(THREAD_ID), 2), + )) as TranscriptStreamItem[]; + expect(items[0]?.kind).toBe("snapshot"); + expect(items[0]?.records.map((r) => r.uuid)).toEqual(["u1"]); + expect(items[1]?.kind).toBe("append"); + expect(items[1]?.records.map((r) => r.uuid)).toEqual(["a1"]); + }).pipe(Effect.scoped, Effect.provide(TestLayer)), +); diff --git a/apps/server/tests/ru-fork/qwen-transcript/cliSettings.test.ts b/apps/server/tests/ru-fork/qwen-transcript/cliSettings.test.ts new file mode 100644 index 00000000000..435545d7257 --- /dev/null +++ b/apps/server/tests/ru-fork/qwen-transcript/cliSettings.test.ts @@ -0,0 +1,94 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { describe, expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; + +import { + pickRuntimeOutputDir, + readRuntimeOutputDirOverride, +} from "../../../src/ru-fork/qwen-transcript/cliSettings.ts"; + +describe("pickRuntimeOutputDir", () => { + it("returns the string value", () => { + expect(pickRuntimeOutputDir({ advanced: { runtimeOutputDir: "/x" } })).toBe("/x"); + }); + it("returns undefined when advanced is missing", () => { + expect(pickRuntimeOutputDir({})).toBeUndefined(); + }); + it("returns undefined when runtimeOutputDir is missing", () => { + expect(pickRuntimeOutputDir({ advanced: {} })).toBeUndefined(); + }); + it("returns undefined for non-object input", () => { + expect(pickRuntimeOutputDir(null)).toBeUndefined(); + expect(pickRuntimeOutputDir("str")).toBeUndefined(); + expect(pickRuntimeOutputDir({ advanced: 5 })).toBeUndefined(); + }); + it("returns undefined for an empty/whitespace value", () => { + expect(pickRuntimeOutputDir({ advanced: { runtimeOutputDir: " " } })).toBeUndefined(); + }); +}); + +const writeSettings = (dir: string, content: string) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + yield* fs.makeDirectory(dir, { recursive: true }); + yield* fs.writeFileString(path.join(dir, "settings.json"), content); + }); + +it.layer(NodeServices.layer)("readRuntimeOutputDirOverride", (it) => { + const setup = Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "t3-cli-settings-" }); + const cliConfigDir = path.join(root, "global"); + const cwd = path.join(root, "cwd"); + return { fs, path, cliConfigDir, cwd }; + }); + + it.effect("reads the global setting", () => + Effect.gen(function* () { + const { fs, path, cliConfigDir, cwd } = yield* setup; + yield* writeSettings(cliConfigDir, JSON.stringify({ advanced: { runtimeOutputDir: "/g" } })); + const result = yield* readRuntimeOutputDirOverride(fs, path, cliConfigDir, cwd); + expect(result).toBe("/g"); + }), + ); + + it.effect("reads the workspace setting", () => + Effect.gen(function* () { + const { fs, path, cliConfigDir, cwd } = yield* setup; + yield* writeSettings(path.join(cwd, ".qwen"), JSON.stringify({ advanced: { runtimeOutputDir: "/w" } })); + const result = yield* readRuntimeOutputDirOverride(fs, path, cliConfigDir, cwd); + expect(result).toBe("/w"); + }), + ); + + it.effect("workspace overrides global", () => + Effect.gen(function* () { + const { fs, path, cliConfigDir, cwd } = yield* setup; + yield* writeSettings(cliConfigDir, JSON.stringify({ advanced: { runtimeOutputDir: "/g" } })); + yield* writeSettings(path.join(cwd, ".qwen"), JSON.stringify({ advanced: { runtimeOutputDir: "/w" } })); + const result = yield* readRuntimeOutputDirOverride(fs, path, cliConfigDir, cwd); + expect(result).toBe("/w"); + }), + ); + + it.effect("returns undefined when neither file exists", () => + Effect.gen(function* () { + const { fs, path, cliConfigDir, cwd } = yield* setup; + const result = yield* readRuntimeOutputDirOverride(fs, path, cliConfigDir, cwd); + expect(result).toBeUndefined(); + }), + ); + + it.effect("treats malformed JSON as undefined", () => + Effect.gen(function* () { + const { fs, path, cliConfigDir, cwd } = yield* setup; + yield* writeSettings(cliConfigDir, "{not json"); + const result = yield* readRuntimeOutputDirOverride(fs, path, cliConfigDir, cwd); + expect(result).toBeUndefined(); + }), + ); +}); diff --git a/apps/server/tests/ru-fork/qwen-transcript/parse.test.ts b/apps/server/tests/ru-fork/qwen-transcript/parse.test.ts new file mode 100644 index 00000000000..0c20efb7cf2 --- /dev/null +++ b/apps/server/tests/ru-fork/qwen-transcript/parse.test.ts @@ -0,0 +1,373 @@ +import { describe, expect, it } from "vitest"; + +import { + normalizeRecord, + parseAndNormalize, + parseTranscriptJsonl, +} from "../../../src/ru-fork/qwen-transcript/parse.ts"; + +const baseRow = (over: Record): Record => ({ + uuid: "u1", + parentUuid: null, + sessionId: "s1", + timestamp: "2026-01-01T00:00:00.000Z", + cwd: "/work", + ...over, +}); + +describe("parseTranscriptJsonl", () => { + it("parses one object per line and skips blank lines", () => { + const text = '{"a":1}\n\n \n{"b":2}\n'; + const { rows, errorCount } = parseTranscriptJsonl(text); + expect(rows).toEqual([{ a: 1 }, { b: 2 }]); + expect(errorCount).toBe(0); + }); + + it("counts malformed lines and skips them", () => { + const text = '{"a":1}\n{not json}\n{"b":2}'; + const { rows, errorCount } = parseTranscriptJsonl(text); + expect(rows).toEqual([{ a: 1 }, { b: 2 }]); + expect(errorCount).toBe(1); + }); + + it("counts non-object JSON values as errors", () => { + const { rows, errorCount } = parseTranscriptJsonl("123\n[1,2]\n{\"ok\":true}"); + expect(rows).toEqual([{ ok: true }]); + expect(errorCount).toBe(2); + }); + + it("returns empty for empty input", () => { + expect(parseTranscriptJsonl("")).toEqual({ rows: [], errorCount: 0 }); + }); +}); + +describe("normalizeRecord — base validation", () => { + it("returns null without uuid/sessionId/timestamp", () => { + expect(normalizeRecord({ type: "user" })).toBeNull(); + expect(normalizeRecord({ type: "user", uuid: "u" })).toBeNull(); + expect(normalizeRecord({ type: "user", uuid: "u", sessionId: "s" })).toBeNull(); + }); + + it("returns null for an unknown type", () => { + expect(normalizeRecord(baseRow({ type: "mystery" }))).toBeNull(); + }); + + it("keeps a string parentUuid and defaults non-strings to null", () => { + const withParent = normalizeRecord(baseRow({ type: "user", parentUuid: "p1" })); + expect(withParent?.parentUuid).toBe("p1"); + const noParent = normalizeRecord(baseRow({ type: "user", parentUuid: 42 })); + expect(noParent?.parentUuid).toBeNull(); + }); + + it("carries gitBranch only when present", () => { + expect(normalizeRecord(baseRow({ type: "user", gitBranch: "main" }))).toMatchObject({ + gitBranch: "main", + }); + expect(normalizeRecord(baseRow({ type: "user" }))).not.toHaveProperty("gitBranch"); + }); +}); + +describe("normalizeRecord — parts", () => { + it("normalizes text / thought / function_call / function_response / inline_data / unknown", () => { + const record = normalizeRecord( + baseRow({ + type: "user", + message: { + parts: [ + { text: "hello" }, + { thought: true, text: "thinking" }, + { functionCall: { name: "Edit", args: { file: "a" } } }, + { functionResponse: { name: "Edit", response: { ok: true } } }, + { inlineData: { mimeType: "image/png" } }, + { somethingElse: 1 }, + "raw-string", + ], + }, + }), + ); + expect(record?.type).toBe("user"); + expect(record && "parts" in record ? record.parts : []).toEqual([ + { kind: "text", text: "hello" }, + { kind: "thought", text: "thinking" }, + { kind: "function_call", name: "Edit", args: { file: "a" } }, + { kind: "function_response", name: "Edit", response: { ok: true } }, + { kind: "inline_data", mimeType: "image/png" }, + { kind: "unknown", raw: { somethingElse: 1 } }, + { kind: "unknown", raw: "raw-string" }, + ]); + }); + + it("returns [] parts when message is missing or parts is not an array", () => { + expect(normalizeRecord(baseRow({ type: "user" }))).toMatchObject({ parts: [] }); + expect(normalizeRecord(baseRow({ type: "user", message: { parts: "nope" } }))).toMatchObject({ + parts: [], + }); + }); + + it("defaults a thought part with no text to empty string", () => { + const record = normalizeRecord( + baseRow({ type: "user", message: { parts: [{ thought: true }] } }), + ); + expect(record && "parts" in record ? record.parts : []).toEqual([{ kind: "thought", text: "" }]); + }); +}); + +describe("normalizeRecord — assistant", () => { + it("maps model, contextWindowSize and usage", () => { + const record = normalizeRecord( + baseRow({ + type: "assistant", + model: "claude-opus-4", + contextWindowSize: 200000, + usageMetadata: { + promptTokenCount: 10, + cachedInputTokenCount: 2, + candidatesTokenCount: 5, + totalTokenCount: 17, + }, + }), + ); + expect(record).toMatchObject({ + type: "assistant", + model: "claude-opus-4", + contextWindowSize: 200000, + usage: { promptTokens: 10, cachedTokens: 2, outputTokens: 5, totalTokens: 17 }, + }); + }); + + it("omits usage when usageMetadata is absent or empty", () => { + expect(normalizeRecord(baseRow({ type: "assistant" }))).not.toHaveProperty("usage"); + expect(normalizeRecord(baseRow({ type: "assistant", usageMetadata: {} }))).not.toHaveProperty( + "usage", + ); + }); +}); + +describe("normalizeRecord — tool_result display variants", () => { + const toolResult = (resultDisplay: unknown, extra: Record = {}) => + normalizeRecord( + baseRow({ type: "tool_result", toolCallResult: { callId: "c1", status: "success", resultDisplay, ...extra } }), + ); + + it("string display → text", () => { + expect(toolResult("done")).toMatchObject({ + toolCall: { callId: "c1", status: "success", display: { kind: "text", text: "done" } }, + }); + }); + + it("file_diff display", () => { + expect( + toolResult({ + fileName: "a.ts", + fileDiff: "@@", + originalContent: "old", + newContent: "new", + diffStat: { added: 1 }, + }), + ).toMatchObject({ + toolCall: { + display: { + kind: "file_diff", + fileName: "a.ts", + fileDiff: "@@", + originalContent: "old", + newContent: "new", + diffStat: { added: 1 }, + }, + }, + }); + }); + + it("file_diff with non-string originalContent becomes null", () => { + const record = toolResult({ fileName: "a", fileDiff: "d", newContent: "n" }); + const display = record && "toolCall" in record ? record.toolCall?.display : undefined; + expect(display).toMatchObject({ kind: "file_diff", originalContent: null }); + }); + + it("todo_list display (with invalid status defaulting to pending)", () => { + const record = toolResult({ + type: "todo_list", + todos: [ + { id: "1", content: "a", status: "completed" }, + { id: "2", content: "b", status: "bogus" }, + "not-an-object", + ], + }); + const display = record && "toolCall" in record ? record.toolCall?.display : undefined; + expect(display).toEqual({ + kind: "todo_list", + todos: [ + { id: "1", content: "a", status: "completed" }, + { id: "2", content: "b", status: "pending" }, + ], + }); + }); + + it("plan_summary display (+ rejected)", () => { + expect( + toolResult({ type: "plan_summary", message: "m", plan: "p", rejected: true }), + ).toMatchObject({ toolCall: { display: { kind: "plan_summary", message: "m", plan: "p", rejected: true } } }); + }); + + it("task_execution display (+ result + toolCalls)", () => { + expect( + toolResult({ + type: "task_execution", + subagentName: "explorer", + status: "completed", + taskDescription: "look", + result: "found", + toolCalls: [{ callId: "x" }], + }), + ).toMatchObject({ + toolCall: { + display: { + kind: "task_execution", + subagentName: "explorer", + status: "completed", + taskDescription: "look", + result: "found", + toolCalls: [{ callId: "x" }], + }, + }, + }); + }); + + it("mcp_tool_progress display (+ total + message)", () => { + expect( + toolResult({ type: "mcp_tool_progress", progress: 3, total: 10, message: "half" }), + ).toMatchObject({ + toolCall: { display: { kind: "mcp_progress", progress: 3, total: 10, message: "half" } }, + }); + }); + + it("ansiOutput display → ansi", () => { + expect(toolResult({ ansiOutput: [["x"]] })).toMatchObject({ + toolCall: { display: { kind: "ansi", raw: [["x"]] } }, + }); + }); + + it("unrecognized object display → unknown", () => { + expect(toolResult({ type: "future_kind" })).toMatchObject({ + toolCall: { display: { kind: "unknown" } }, + }); + }); + + it("drops an invalid status and omits display when absent", () => { + const record = normalizeRecord( + baseRow({ type: "tool_result", toolCallResult: { callId: "c", status: "bogus" } }), + ); + const toolCall = record && "toolCall" in record ? record.toolCall : undefined; + expect(toolCall).toEqual({ callId: "c" }); + }); + + it("omits toolCall entirely when toolCallResult is absent", () => { + expect(normalizeRecord(baseRow({ type: "tool_result" }))).not.toHaveProperty("toolCall"); + }); +}); + +describe("normalizeRecord — system", () => { + it("keeps a known subtype and payload", () => { + expect( + normalizeRecord( + baseRow({ type: "system", subtype: "chat_compression", systemPayload: { info: 1 } }), + ), + ).toMatchObject({ type: "system", subtype: "chat_compression", payload: { info: 1 } }); + }); + + it("drops an unknown subtype but stays a system record", () => { + const record = normalizeRecord(baseRow({ type: "system", subtype: "future" })); + expect(record).toMatchObject({ type: "system" }); + expect(record).not.toHaveProperty("subtype"); + }); +}); + +describe("normalizeRecord — defensive defaults for missing fields", () => { + const display = (resultDisplay: unknown) => { + const record = normalizeRecord(baseRow({ type: "tool_result", toolCallResult: { resultDisplay } })); + return record && "toolCall" in record ? record.toolCall?.display : undefined; + }; + + it("defaults cwd to empty string when absent", () => { + const record = normalizeRecord({ uuid: "u", sessionId: "s", timestamp: "t", type: "user" }); + expect(record).toMatchObject({ cwd: "" }); + }); + + it("defaults function_call / function_response / inline_data fields when absent", () => { + const record = normalizeRecord( + baseRow({ + type: "user", + message: { parts: [{ functionCall: {} }, { functionResponse: {} }, { inlineData: {} }] }, + }), + ); + expect(record && "parts" in record ? record.parts : []).toEqual([ + { kind: "function_call", name: "", args: undefined }, + { kind: "function_response", name: "", response: undefined }, + { kind: "inline_data", mimeType: "" }, + ]); + }); + + it("plan_summary with no fields → empty strings, no rejected", () => { + expect(display({ type: "plan_summary" })).toEqual({ kind: "plan_summary", message: "", plan: "" }); + }); + + it("task_execution with no optionals → empty strings, no result/toolCalls", () => { + expect(display({ type: "task_execution" })).toEqual({ + kind: "task_execution", + subagentName: "", + status: "", + taskDescription: "", + }); + }); + + it("mcp_tool_progress with no fields → progress 0, no total/message", () => { + expect(display({ type: "mcp_tool_progress" })).toEqual({ kind: "mcp_progress", progress: 0 }); + }); + + it("todo_list with non-array todos → unknown", () => { + expect(display({ type: "todo_list" })).toMatchObject({ kind: "unknown" }); + }); + + it("file_diff with no fileName/newContent → empty strings, null original", () => { + expect(display({ fileDiff: "@@" })).toEqual({ + kind: "file_diff", + fileName: "", + fileDiff: "@@", + originalContent: null, + newContent: "", + }); + }); + + it("todo_list maps the in_progress status", () => { + expect(display({ type: "todo_list", todos: [{ id: "1", content: "c", status: "in_progress" }] })).toEqual({ + kind: "todo_list", + todos: [{ id: "1", content: "c", status: "in_progress" }], + }); + }); + + it("non-object, non-string display → unknown", () => { + expect(display(42)).toEqual({ kind: "unknown", raw: 42 }); + }); + + it("todo item missing id/content → empty strings", () => { + expect(display({ type: "todo_list", todos: [{ status: "completed" }] })).toEqual({ + kind: "todo_list", + todos: [{ id: "", content: "", status: "completed" }], + }); + }); +}); + +describe("parseAndNormalize", () => { + it("parses + normalizes + drops unusable rows in order", () => { + const text = [ + JSON.stringify(baseRow({ uuid: "a", type: "user", message: { parts: [{ text: "hi" }] } })), + JSON.stringify({ type: "user" }), // missing ids → dropped + "{bad json}", // malformed → dropped + JSON.stringify(baseRow({ uuid: "b", type: "assistant", model: "m" })), + ].join("\n"); + const records = parseAndNormalize(text); + expect(records.map((r) => r.uuid)).toEqual(["a", "b"]); + expect(records[0]?.type).toBe("user"); + expect(records[1]?.type).toBe("assistant"); + }); +}); diff --git a/apps/server/tests/ru-fork/qwen-transcript/paths.test.ts b/apps/server/tests/ru-fork/qwen-transcript/paths.test.ts new file mode 100644 index 00000000000..3c1d8bef550 --- /dev/null +++ b/apps/server/tests/ru-fork/qwen-transcript/paths.test.ts @@ -0,0 +1,113 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as os from "node:os"; +import * as path from "node:path"; + +import { describe, expect, it } from "vitest"; + +import { + expandRuntimeBaseDir, + resolveChatsDir, + resolveTranscriptBaseDir, + resolveTranscriptFilePath, + sanitizeCwd, + type TranscriptBaseInput, +} from "../../../src/ru-fork/qwen-transcript/paths.ts"; + +const base = (overrides: Partial): TranscriptBaseInput => ({ + env: {}, + cliConfigDir: "/home/user/.qwen", + cwd: "/work/project", + runtimeOutputDirSetting: undefined, + ...overrides, +}); + +describe("sanitizeCwd", () => { + it("replaces every non-alphanumeric character with a hyphen", () => { + expect(sanitizeCwd("/Users/x/My Project.v2")).toBe("-Users-x-My-Project-v2"); + }); + + it("preserves alphanumerics", () => { + expect(sanitizeCwd("abc123")).toBe("abc123"); + }); + + it("lowercases on win32", () => { + expect(sanitizeCwd("C:\\Foo", "win32")).toBe("c--foo"); + }); +}); + +describe("expandRuntimeBaseDir", () => { + it("expands a bare tilde to the home directory", () => { + expect(expandRuntimeBaseDir("~", "/work")).toBe(os.homedir()); + }); + + it("expands ~/segment relative to home", () => { + expect(expandRuntimeBaseDir("~/foo/bar", "/work")).toBe(path.join(os.homedir(), "foo", "bar")); + }); + + it("expands ~\\segment (backslash) relative to home", () => { + expect(expandRuntimeBaseDir("~\\foo", "/work")).toBe(path.join(os.homedir(), "foo")); + }); + + it("resolves a relative path against cwd", () => { + expect(expandRuntimeBaseDir("out/dir", "/work")).toBe("/work/out/dir"); + }); + + it("leaves an absolute path unchanged", () => { + expect(expandRuntimeBaseDir("/abs/dir", "/work")).toBe("/abs/dir"); + }); +}); + +describe("resolveTranscriptBaseDir", () => { + it("prefers an absolute QWEN_RUNTIME_DIR env var", () => { + expect(resolveTranscriptBaseDir(base({ env: { QWEN_RUNTIME_DIR: "/custom/runtime" } }))).toBe( + "/custom/runtime", + ); + }); + + it("expands a tilde QWEN_RUNTIME_DIR", () => { + expect(resolveTranscriptBaseDir(base({ env: { QWEN_RUNTIME_DIR: "~/rt" } }))).toBe( + path.join(os.homedir(), "rt"), + ); + }); + + it("resolves a relative QWEN_RUNTIME_DIR against cwd", () => { + expect( + resolveTranscriptBaseDir(base({ env: { QWEN_RUNTIME_DIR: "rt" }, cwd: "/c" })), + ).toBe("/c/rt"); + }); + + it("ignores a whitespace-only QWEN_RUNTIME_DIR and falls through", () => { + expect(resolveTranscriptBaseDir(base({ env: { QWEN_RUNTIME_DIR: " " } }))).toBe( + "/home/user/.qwen", + ); + }); + + it("uses the runtimeOutputDir setting when no env var", () => { + expect(resolveTranscriptBaseDir(base({ runtimeOutputDirSetting: "/setting/dir" }))).toBe( + "/setting/dir", + ); + }); + + it("expands a relative runtimeOutputDir setting against cwd", () => { + expect( + resolveTranscriptBaseDir(base({ runtimeOutputDirSetting: "rel", cwd: "/c" })), + ).toBe("/c/rel"); + }); + + it("falls back to cliConfigDir when nothing is set", () => { + expect(resolveTranscriptBaseDir(base({}))).toBe("/home/user/.qwen"); + }); +}); + +describe("resolveChatsDir / resolveTranscriptFilePath", () => { + it("assembles /projects//chats with no tmp segment or hash", () => { + const chatsDir = resolveChatsDir(base({ cwd: "/work/project" })); + expect(chatsDir).toBe("/home/user/.qwen/projects/-work-project/chats"); + expect(chatsDir).not.toContain("/tmp/"); + }); + + it("appends .jsonl", () => { + const filePath = resolveTranscriptFilePath(base({ cwd: "/work/project", sessionId: "abc-123" })); + expect(filePath).toBe("/home/user/.qwen/projects/-work-project/chats/abc-123.jsonl"); + }); +}); diff --git a/apps/web/package.json b/apps/web/package.json index 0bd03ee0bf4..e712d090152 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -30,6 +30,7 @@ "@xterm/addon-fit": "0.11.0", "@xterm/xterm": "6.0.0", "class-variance-authority": "0.7.1", + "diff": "8.0.3", "effect": "catalog:", "lexical": "0.41.0", "lucide-react": "0.564.0", diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 57cad25cd6d..3c3d1b0c9ff 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -155,6 +155,10 @@ import { flattenSubagentBuckets } from "../ru-fork/subagents/composerIntegration // command outside its ACP allowlist (otherwise -32603 "Internal error" crash). import { applyUnknownSlashCommandStripToComposer } from "../ru-fork/unsupported-slash-commands/applyUnknownSlashCommandStripToComposer"; import { ChatHeader } from "./chat/ChatHeader"; +// ru-fork: advanced chat mode (qwen CLI transcript view). +import { AdvancedMessagesTimeline } from "~/ru-fork/advanced-chat/AdvancedMessagesTimeline"; +import { useAdvancedChatMode } from "~/ru-fork/advanced-chat/advancedChatMode"; +import { useTranscriptSubscription } from "~/ru-fork/advanced-chat/useTranscriptSubscription"; import { type ExpandedImagePreview } from "./chat/ExpandedImagePreview"; import { NoActiveThreadState } from "./NoActiveThreadState"; import { resolveEffectiveEnvMode, resolveEnvironmentOptionLabel } from "./BranchToolbar.logic"; @@ -1536,6 +1540,18 @@ export default function ChatView(props: ChatViewProps) { const activeProjectCwd = activeProject?.cwd ?? null; const activeThreadWorktreePath = activeThread?.worktreePath ?? null; const activeWorkspaceRoot = activeThreadWorktreePath ?? activeProjectCwd ?? undefined; + // ru-fork: advanced chat mode — render qwen's CLI transcript instead of the + // projection timeline. Per-thread toggle; transcript streamed read-only. + const advancedChatModeByThread = useAdvancedChatMode((state) => state.byThread); + const toggleAdvancedChatMode = useAdvancedChatMode((state) => state.toggle); + const advancedChatOpen = activeThread + ? Boolean(advancedChatModeByThread[activeThread.id]) + : false; + const transcriptRecords = useTranscriptSubscription( + activeThread?.environmentId, + activeThread?.id, + advancedChatOpen, + ); const activeTerminalLaunchContext = terminalLaunchContext?.threadId === activeThreadId ? terminalLaunchContext @@ -1604,7 +1620,15 @@ export default function ChatView(props: ChatViewProps) { return diffOpen ? { ...rest, diff: undefined } : { ...rest, diff: "1" }; }, }); - }, [diffOpen, environmentId, isServerThread, navigate, onDiffPanelOpen, setMcpPanelOpen, threadId]); + }, [ + diffOpen, + environmentId, + isServerThread, + navigate, + onDiffPanelOpen, + setMcpPanelOpen, + threadId, + ]); const onToggleMcp = useCallback(() => { const next = !mcpPanelOpen; @@ -3611,6 +3635,10 @@ export default function ChatView(props: ChatViewProps) { gitCwd={gitCwd} diffOpen={diffOpen} mcpPanelOpen={mcpPanelOpen} + advancedChatOpen={advancedChatOpen} + onToggleAdvancedChat={() => { + if (activeThread) toggleAdvancedChatMode(activeThread.id); + }} onRunProjectScript={runProjectScript} onAddProjectScript={saveProjectScript} onUpdateProjectScript={updateProjectScript} @@ -3633,36 +3661,48 @@ export default function ChatView(props: ChatViewProps) {
{/* Messages Wrapper */}
- {/* Messages — LegendList handles virtualization and scrolling internally */} - + {/* ru-fork: advanced chat mode swaps the projection timeline for the + read-only qwen CLI transcript view; composer/approvals stay live. */} + {advancedChatOpen ? ( + + ) : ( + <> + {/* Messages — LegendList handles virtualization and scrolling internally */} + + + )} {/* scroll to bottom pill — shown when user has scrolled away from the bottom */} - {showScrollToBottom && ( + {!advancedChatOpen && showScrollToBottom && (
); diff --git a/apps/web/src/rpc/wsRpcClient.ts b/apps/web/src/rpc/wsRpcClient.ts index 57a190cb807..565e1643899 100644 --- a/apps/web/src/rpc/wsRpcClient.ts +++ b/apps/web/src/rpc/wsRpcClient.ts @@ -7,6 +7,7 @@ import { type LocalApi, ORCHESTRATION_WS_METHODS, type ServerSettingsPatch, + TRANSCRIPT_WS_METHOD, WS_METHODS, } from "@t3tools/contracts"; import { applyGitStatusStreamEvent } from "@t3tools/shared/git"; @@ -150,6 +151,8 @@ export interface WsRpcClient { >; readonly subscribeShell: RpcStreamMethod; readonly subscribeThread: RpcInputStreamMethod; + // ru-fork: advanced chat mode — qwen transcript stream. + readonly subscribeTranscript: RpcInputStreamMethod; }; // ru-fork: MCP catalog/bindings reads + runtime status. Mutations go through // `orchestration.dispatchCommand` (the 5 mcp.* commands). @@ -317,6 +320,12 @@ export function createWsRpcClient(transport: WsTransport): WsRpcClient { listener, { ...options, tag: ORCHESTRATION_WS_METHODS.subscribeThread }, ), + // ru-fork: advanced chat mode — qwen transcript stream. + subscribeTranscript: (input, listener, options) => + transport.subscribe((client) => client[TRANSCRIPT_WS_METHOD](input), listener, { + ...options, + tag: TRANSCRIPT_WS_METHOD, + }), }, mcp: { getSnapshot: (input) => diff --git a/apps/web/src/ru-fork/advanced-chat/AdvancedMessagesTimeline.tsx b/apps/web/src/ru-fork/advanced-chat/AdvancedMessagesTimeline.tsx new file mode 100644 index 00000000000..57207d65dbb --- /dev/null +++ b/apps/web/src/ru-fork/advanced-chat/AdvancedMessagesTimeline.tsx @@ -0,0 +1,124 @@ +// ru-fork: advanced chat mode timeline. Renders the merged transcript flow: +// user bubbles, assistant text (no redundant label) with model/token/latency +// badges, merged tool-step cards (call + result + telemetry), and system +// dividers. Theme-safe semantic tokens; heavy content collapses, nothing clamps. +import type { TranscriptRecord } from "@t3tools/contracts"; +import { InfoIcon } from "lucide-react"; + +import { Badge } from "~/components/ui/badge"; + +import { ToolStepCard } from "./ToolStep"; +import { TranscriptParts } from "./TranscriptParts"; +import { UserBubble } from "./UserBubble"; +import { buildTranscriptFlow, type AssistantFlowItem } from "./transcriptFlow"; +import type { SystemRecord } from "./transcriptTypes"; + +const SYSTEM_LABELS: Record, string> = { + chat_compression: "Контекст сжат", + slash_command: "Slash-команда", + ui_telemetry: "Телеметрия", + at_command: "@-команда", +}; + +function formatLatency(ms: number | undefined): string | null { + if (ms === undefined) return null; + return ms < 1000 ? `${ms} мс` : `${(ms / 1000).toFixed(1)} с`; +} + +function AssistantTurn({ item, cwd }: { item: AssistantFlowItem; cwd: string | undefined }) { + const { record } = item; + // function_call parts become tool steps; everything else renders as prose. + const textParts = record.parts.filter((part) => part.kind !== "function_call"); + const usage = record.usage; + const latency = formatLatency(item.latencyMs); + const tokenBadges: Array<[string, number | undefined]> = [ + ["вход", usage?.promptTokens], + ["кэш", usage?.cachedTokens], + ["выход", usage?.outputTokens], + ["Σ", usage?.totalTokens], + ]; + + return ( +
+ {textParts.length > 0 ? ( +
+ +
+ ) : null} +
+ {record.model ? ( + + {record.model} + + ) : null} + {tokenBadges.map(([label, value]) => + value === undefined ? null : ( + + {label} {value} + + ), + )} + {latency ? ( + + {latency} + + ) : null} +
+ {item.tools.map((step) => ( + + ))} +
+ ); +} + +function SystemDivider({ record }: { record: SystemRecord }) { + const label = record.subtype ? SYSTEM_LABELS[record.subtype] : "Системное событие"; + const date = new Date(record.timestamp); + const time = Number.isNaN(date.getTime()) ? "" : date.toLocaleTimeString(); + return ( +
+ + + {label} + {time ? ( + + {time} + + ) : null} + +
+ ); +} + +export function AdvancedMessagesTimeline({ + records, + markdownCwd, +}: { + records: ReadonlyArray; + markdownCwd: string | undefined; +}) { + if (records.length === 0) { + return ( +
+ Пока нет записей в стенограмме CLI для этого диалога. +
+ ); + } + const flow = buildTranscriptFlow(records); + return ( +
+
+ {flow.map((item) => { + switch (item.kind) { + case "user": + return ; + case "assistant": + return ; + case "system": + return ; + } + })} +
+
+ ); +} diff --git a/apps/web/src/ru-fork/advanced-chat/ToolStep.tsx b/apps/web/src/ru-fork/advanced-chat/ToolStep.tsx new file mode 100644 index 00000000000..f88f6540bac --- /dev/null +++ b/apps/web/src/ru-fork/advanced-chat/ToolStep.tsx @@ -0,0 +1,522 @@ +// ru-fork: advanced chat mode — one merged tool step (call + result + telemetry). +// Header: tool name · status · duration. Body: command/args, diff (collapsed, +// with +/- diffStat), result variants, and a visible reason for cancel/error. +import { getFiletypeFromFileName, parsePatchFiles } from "@pierre/diffs"; +import { FileDiff } from "@pierre/diffs/react"; +import type { TranscriptPart, TranscriptToolDisplay } from "@t3tools/contracts"; +import { createTwoFilesPatch } from "diff"; +import { + AlertTriangleIcon, + CheckCircle2Icon, + ChevronDownIcon, + ChevronRightIcon, + CircleDashedIcon, + CircleIcon, + TerminalIcon, + WrenchIcon, +} from "lucide-react"; +import { type ReactNode, useMemo, useState } from "react"; + +import ChatMarkdown from "~/components/ChatMarkdown"; +import { Badge } from "~/components/ui/badge"; +import { useTheme } from "~/hooks/useTheme"; +import { resolveDiffThemeName } from "~/lib/diffRendering"; +import { cn } from "~/lib/utils"; +import { ADVANCED_CHAT_PENDING_PREVIEW, ADVANCED_CHAT_RICH_DIFF } from "~/ru-fork/config"; + +import { Disclosure, JsonBlock } from "./transcriptVisuals"; +import type { ToolStep, ToolStepStatus } from "./transcriptFlow"; + +type BadgeVariant = "secondary" | "success" | "error" | "info"; + +const STATUS_META: Record = { + success: { label: "успешно", variant: "success" }, + error: { label: "ошибка", variant: "error" }, + cancelled: { label: "отменено", variant: "secondary" }, + running: { label: "выполняется", variant: "info" }, +}; + +function formatDuration(ms: number | undefined): string | null { + if (ms === undefined) return null; + return ms < 1000 ? `${ms} мс` : `${(ms / 1000).toFixed(1)} с`; +} + +function diffStatLabel(diffStat: unknown): ReactNode { + if (typeof diffStat !== "object" || diffStat === null) return null; + const stat = diffStat as Record; + const added = typeof stat["model_added_lines"] === "number" ? stat["model_added_lines"] : 0; + const removed = typeof stat["model_removed_lines"] === "number" ? stat["model_removed_lines"] : 0; + if (added === 0 && removed === 0) return null; + return ( + + +{added} + −{removed} + + ); +} + +/** Raw fallback: the unified patch text, Shiki-highlighted, behind a disclosure. */ +function RawDiffView({ + patch, + fileName, + diffStat, + cwd, +}: { + patch: string; + fileName: string; + diffStat: unknown; + cwd: string | undefined; +}) { + return ( + {fileName || "diff"}} + meta={ + diffStatLabel(diffStat) ?? ( + + diff + + ) + } + > + + + ); +} + +/** Rich renderer: parse the unified patch and render @pierre/diffs FileDiff + * (clean gutter, native collapse, theme-aware). Falls back to RawDiffView when + * the patch can't be parsed. The diff worker pool is provided by the route. */ +function FileDiffView({ + patch, + fileName, + diffStat, + cwd, +}: { + patch: string; + fileName: string; + diffStat: unknown; + cwd: string | undefined; +}) { + const { resolvedTheme } = useTheme(); + const [collapsed, setCollapsed] = useState(true); + const files = useMemo(() => { + const trimmed = patch.trim(); + if (trimmed.length === 0) return null; + try { + const parsed = parsePatchFiles(trimmed, `transcript:${fileName}:${trimmed.length}`).flatMap( + (entry) => entry.files, + ); + return parsed.length > 0 ? parsed : null; + } catch { + return null; + } + }, [patch, fileName]); + + if (!files) { + return ; + } + + return ( +
+ {files.map((fileDiff, index) => ( + ( + + )} + options={{ + collapsed, + diffStyle: "unified", + lineDiffType: "none", + overflow: "wrap", + theme: resolveDiffThemeName(resolvedTheme), + themeType: resolvedTheme, + }} + /> + ))} +
+ ); +} + +/** Render a unified patch with the rich renderer or the raw fallback per flag. */ +function DiffBody(props: { + patch: string; + fileName: string; + diffStat: unknown; + cwd: string | undefined; +}) { + return ADVANCED_CHAT_RICH_DIFF ? : ; +} + +const TODO_ICON: Record = { + completed: , + in_progress: , + pending: , +}; + +function DisplayView({ + display, + cwd, +}: { + display: TranscriptToolDisplay; + cwd: string | undefined; +}) { + switch (display.kind) { + case "text": + return display.text.trim().length === 0 ? null : ( +
+          {display.text}
+        
+ ); + case "file_diff": + return ( + + ); + case "todo_list": + return ( +
    + {display.todos.map((todo) => ( +
  • + + {TODO_ICON[todo.status] ?? TODO_ICON["pending"]} + + + {todo.content} + +
  • + ))} +
+ ); + case "plan_summary": + return ( +
+ {display.rejected ? ( + + план отклонён + + ) : null} + {display.message.trim().length > 0 ? ( + + ) : null} + {display.plan.trim().length > 0 ? ( +
+ +
+ ) : null} +
+ ); + case "task_execution": + return ( +
+
+ + {display.subagentName || "подагент"} + + + {display.status} + +
+ {display.taskDescription.trim().length > 0 ? ( +

{display.taskDescription}

+ ) : null} + {display.result && display.result.trim().length > 0 ? ( + + ) : null} + {display.toolCalls && display.toolCalls.length > 0 ? ( + + {display.toolCalls.length} + + } + > + + + ) : null} +
+ ); + case "mcp_progress": { + const ratio = + display.total && display.total > 0 + ? Math.min(1, Math.max(0, display.progress / display.total)) + : null; + return ( +
+
+ {display.message ?? "выполнение"} + + {display.progress} + {display.total !== undefined ? ` / ${display.total}` : ""} + +
+ {ratio !== null ? ( +
+
+
+ ) : null} +
+ ); + } + case "ansi": + return ( + + + + ); + case "unknown": + return ( + + raw + + } + > + + + ); + } +} + +function responsePreview( + parts: ReadonlyArray, +): { preview: string; value: unknown } | null { + for (const part of parts) { + if (part.kind !== "function_response") continue; + const response = part.response; + const output = + typeof response === "string" + ? response + : typeof response === "object" && + response !== null && + typeof (response as Record)["output"] === "string" + ? ((response as Record)["output"] as string) + : undefined; + const preview = (output ?? JSON.stringify(response) ?? "").replace(/\s+/g, " ").trim(); + return { + preview: preview.length > 80 ? `${preview.slice(0, 80)}…` : preview, + value: output ?? response, + }; + } + return null; +} + +type FilePreview = + | { + readonly kind: "diff"; + readonly fileName: string; + readonly patch: string; + readonly replaceAll: boolean; + } + | { + readonly kind: "content"; + readonly fileName: string; + readonly content: string; + readonly language: string; + }; + +const asText = (value: unknown): string | undefined => + typeof value === "string" ? value : undefined; + +function baseName(filePath: string): string { + const segments = filePath.split(/[\\/]/); + return segments[segments.length - 1] || filePath; +} + +/** Wrap `content` in a fence longer than any backtick run inside it, so a file + * that itself contains ``` blocks (e.g. a Markdown doc) can't close it early. */ +function codeFence(content: string, language: string): string { + let longest = 0; + for (const run of content.matchAll(/`+/g)) { + longest = Math.max(longest, run[0].length); + } + const ticks = "`".repeat(Math.max(3, longest + 1)); + return `${ticks}${language}\n${content}\n${ticks}`; +} + +/** A pending edit/write_file — recorded function_call with no result yet — gets + * a preview of what WILL be applied. `edit`/`replace` (qwen ToolNames: `edit`, + * legacy alias `replace`) → an old→new diff synthesized from the call args; + * `write_file` → the proposed content as a code block. null for everything else + * and for already-completed steps (DisplayView owns those). */ +function buildFilePreview(step: ToolStep): FilePreview | null { + if (step.display !== undefined) return null; + if (typeof step.args !== "object" || step.args === null) return null; + const args = step.args as Record; + const filePath = asText(args["file_path"]) ?? ""; + const fileName = filePath ? baseName(filePath) : "файл"; + + if (step.name === "edit" || step.name === "replace") { + const before = asText(args["old_string"]) ?? ""; + const after = asText(args["new_string"]) ?? ""; + const patch = createTwoFilesPatch(fileName, fileName, before, after); + return { kind: "diff", fileName, patch, replaceAll: args["replace_all"] === true }; + } + if (step.name === "write_file") { + const content = asText(args["content"]); + if (content === undefined) return null; + return { kind: "content", fileName, content, language: getFiletypeFromFileName(fileName) }; + } + return null; +} + +const HEAVY_ARG_KEYS: ReadonlySet = new Set(["content", "old_string", "new_string"]); + +/** Drop the large text fields from edit/write_file args so the JSON disclosure + * shows only the small ones (file_path, replace_all); the body/diff is rendered + * on its own, never as escaped JSON. Other tools keep their full args. */ +function compactArgs(name: string, args: unknown): unknown { + if (typeof args !== "object" || args === null) return args; + if (name !== "edit" && name !== "replace" && name !== "write_file") return args; + const compact: Record = {}; + for (const [key, value] of Object.entries(args as Record)) { + if (!HEAVY_ARG_KEYS.has(key)) compact[key] = value; + } + return compact; +} + +function isRenderableArgs(args: unknown): boolean { + if (args === undefined) return false; + if (typeof args === "object" && args !== null) return Object.keys(args).length > 0; + return true; +} + +function FilePreviewView({ preview, cwd }: { preview: FilePreview; cwd: string | undefined }) { + return ( +
+
+ + предпросмотр + + + {preview.fileName} + + {preview.kind === "diff" && preview.replaceAll ? ( + + ко всем совпадениям + + ) : null} +
+ {preview.kind === "diff" ? ( + + ) : ( + + {`${preview.content.split("\n").length} стр.`} + + } + > + + + )} +
+ ); +} + +export function ToolStepCard({ step, cwd }: { step: ToolStep; cwd: string | undefined }) { + const status = STATUS_META[step.status]; + const duration = formatDuration(step.durationMs); + const response = responsePreview(step.response); + const filePreview = useMemo( + () => (ADVANCED_CHAT_PENDING_PREVIEW ? buildFilePreview(step) : null), + [step], + ); + const argsForDisplay = useMemo(() => compactArgs(step.name, step.args), [step.name, step.args]); + + return ( +
+
+ {step.command !== undefined ? ( + + ) : ( + + )} + + {step.name || "инструмент"} + + + {status.label} + + {duration ? ( + {duration} + ) : null} +
+ +
+ {step.command !== undefined ? ( +
+            $ 
+            {step.command}
+          
+ ) : isRenderableArgs(argsForDisplay) ? ( + + + + ) : null} + + {filePreview ? : null} + + {step.error ? ( +
+ + {step.error} +
+ ) : null} + + {step.display ? : null} + + {response && response.preview.length > 0 ? ( + + {response.preview} + + } + > + + + ) : null} +
+
+ ); +} diff --git a/apps/web/src/ru-fork/advanced-chat/TranscriptParts.tsx b/apps/web/src/ru-fork/advanced-chat/TranscriptParts.tsx new file mode 100644 index 00000000000..642d6baea58 --- /dev/null +++ b/apps/web/src/ru-fork/advanced-chat/TranscriptParts.tsx @@ -0,0 +1,116 @@ +// ru-fork: advanced chat mode — renders message `parts` (the @google/genai parts, +// normalized). Every kind is presented: text/thought as prose, calls/responses and +// unknown blobs in collapsibles, inline data as a labelled chip. +import type { TranscriptPart } from "@t3tools/contracts"; +import { BrainIcon, PaperclipIcon } from "lucide-react"; + +import ChatMarkdown from "~/components/ChatMarkdown"; +import { Badge } from "~/components/ui/badge"; + +import { Disclosure, JsonBlock } from "./transcriptVisuals"; + +function PartView({ part, cwd }: { part: TranscriptPart; cwd: string | undefined }) { + switch (part.kind) { + case "text": + return part.text.trim().length === 0 ? null : ; + case "thought": + return ( + + + Размышления + + } + > + + + ); + case "function_call": + return ( + + Вызов инструмента + + {part.name || "—"} + + + } + meta={ + + аргументы + + } + > + + + ); + case "function_response": + return ( + + Ответ инструмента + + {part.name || "—"} + + + } + > + + + ); + case "inline_data": + return ( + + + {part.mimeType || "вложение"} + + ); + case "unknown": + return ( + + raw + + } + > + + + ); + } +} + +// Parts are immutable for a given record, but build a content-derived key so we +// never rely on the bare array index. +function partKey(part: TranscriptPart, index: number): string { + switch (part.kind) { + case "text": + case "thought": + return `${part.kind}-${index}-${part.text.length}`; + case "function_call": + case "function_response": + return `${part.kind}-${index}-${part.name}`; + case "inline_data": + return `inline-${index}-${part.mimeType}`; + case "unknown": + return `unknown-${index}`; + } +} + +export function TranscriptParts({ + parts, + cwd, +}: { + parts: ReadonlyArray; + cwd: string | undefined; +}) { + const rendered = parts.map((part, index) => { + const node = ; + return node ?
{node}
: null; + }); + return
{rendered}
; +} diff --git a/apps/web/src/ru-fork/advanced-chat/UserBubble.tsx b/apps/web/src/ru-fork/advanced-chat/UserBubble.tsx new file mode 100644 index 00000000000..b5f39d650d6 --- /dev/null +++ b/apps/web/src/ru-fork/advanced-chat/UserBubble.tsx @@ -0,0 +1,43 @@ +// ru-fork: advanced chat mode — user message bubble. Matches the existing chat +// bubble look (right-aligned secondary bubble + timestamp + copy); an image part +// shows as a 📎 chip (the transcript carries only the mime type, not the bytes). +import { PaperclipIcon } from "lucide-react"; + +import ChatMarkdown from "~/components/ChatMarkdown"; +import { Badge } from "~/components/ui/badge"; + +import type { UserRecord } from "./transcriptTypes"; +import { CopyButton } from "./transcriptVisuals"; + +export function UserBubble({ record, cwd }: { record: UserRecord; cwd: string | undefined }) { + const text = record.parts + .flatMap((part) => (part.kind === "text" ? [part.text] : [])) + .join("\n\n"); + const images = record.parts.flatMap((part) => + part.kind === "inline_data" ? [part.mimeType] : [], + ); + const date = new Date(record.timestamp); + const time = Number.isNaN(date.getTime()) ? record.timestamp : date.toLocaleTimeString(); + + return ( +
+
+ {text.length > 0 ? : null} + {images.length > 0 ? ( +
+ {images.map((mimeType, index) => ( + + + {mimeType || "вложение"} + + ))} +
+ ) : null} +
+ {text.length > 0 ? : null} + {time} +
+
+
+ ); +} diff --git a/apps/web/src/ru-fork/advanced-chat/advancedChatMode.ts b/apps/web/src/ru-fork/advanced-chat/advancedChatMode.ts new file mode 100644 index 00000000000..e34c3d8e5c2 --- /dev/null +++ b/apps/web/src/ru-fork/advanced-chat/advancedChatMode.ts @@ -0,0 +1,40 @@ +// ru-fork: advanced chat mode toggle — per-thread, persisted to localStorage so the +// view choice survives reloads. Isolated store; does not touch uiStateStore. +import { create } from "zustand"; + +const STORAGE_KEY = "ru-fork:advanced-chat-mode:v1"; + +function loadInitial(): Record { + if (typeof window === "undefined") return {}; + try { + const raw = window.localStorage.getItem(STORAGE_KEY); + const parsed = raw ? (JSON.parse(raw) as unknown) : {}; + return typeof parsed === "object" && parsed !== null ? (parsed as Record) : {}; + } catch { + return {}; + } +} + +function persist(byThread: Record): void { + if (typeof window === "undefined") return; + try { + window.localStorage.setItem(STORAGE_KEY, JSON.stringify(byThread)); + } catch { + // ignore quota / serialization errors + } +} + +interface AdvancedChatModeStore { + readonly byThread: Record; + readonly toggle: (threadId: string) => void; +} + +export const useAdvancedChatMode = create((set) => ({ + byThread: loadInitial(), + toggle: (threadId) => + set((state) => { + const next = { ...state.byThread, [threadId]: !state.byThread[threadId] }; + persist(next); + return { byThread: next }; + }), +})); diff --git a/apps/web/src/ru-fork/advanced-chat/transcriptFlow.ts b/apps/web/src/ru-fork/advanced-chat/transcriptFlow.ts new file mode 100644 index 00000000000..a9b39a6fb15 --- /dev/null +++ b/apps/web/src/ru-fork/advanced-chat/transcriptFlow.ts @@ -0,0 +1,205 @@ +// ru-fork: advanced chat mode — flatten qwen's record stream into a render-ready +// flow. One logical tool use is spread across THREE records (assistant +// functionCall + tool_result + ui_telemetry/tool_call); we merge them into a +// single ToolStep so the UI shows one card with args/diff/status/duration — +// including cancelled tools, which have NO tool_result (only a tool_call event). +// `api_response` telemetry is folded into the assistant line's latency. +import type { TranscriptPart, TranscriptToolDisplay } from "@t3tools/contracts"; + +import type { + AssistantRecord, + SystemRecord, + ToolResultRecord, + UserRecord, +} from "./transcriptTypes"; + +export type ToolStepStatus = "success" | "error" | "cancelled" | "running"; + +export interface ToolStep { + readonly key: string; + name: string; + args: unknown; + command: string | undefined; + status: ToolStepStatus; + durationMs: number | undefined; + error: string | undefined; + display: TranscriptToolDisplay | undefined; + response: ReadonlyArray; +} + +export interface AssistantFlowItem { + readonly kind: "assistant"; + readonly record: AssistantRecord; + latencyMs: number | undefined; + readonly tools: ToolStep[]; +} + +export type FlowItem = + | { readonly kind: "user"; readonly record: UserRecord } + | AssistantFlowItem + | { readonly kind: "system"; readonly record: SystemRecord }; + +interface TelemetryEvent { + readonly name: string; + readonly durationMs: number | undefined; + readonly status: string | undefined; + readonly error: string | undefined; + readonly functionName: string | undefined; + readonly functionArgs: unknown; +} + +const asString = (value: unknown): string | undefined => + typeof value === "string" ? value : undefined; +const asNumber = (value: unknown): number | undefined => + typeof value === "number" && Number.isFinite(value) ? value : undefined; + +function readTelemetry(record: SystemRecord): TelemetryEvent | null { + if (record.subtype !== "ui_telemetry") return null; + const payload = record.payload; + if (typeof payload !== "object" || payload === null) return null; + const event = (payload as Record)["uiEvent"]; + if (typeof event !== "object" || event === null) return null; + const ev = event as Record; + return { + name: asString(ev["event.name"]) ?? "", + durationMs: asNumber(ev["duration_ms"]), + status: asString(ev["status"]), + error: asString(ev["error"]), + functionName: asString(ev["function_name"]), + functionArgs: ev["function_args"], + }; +} + +function shellCommand(name: string, args: unknown): string | undefined { + if (name !== "run_shell_command" || typeof args !== "object" || args === null) return undefined; + return asString((args as Record)["command"]); +} + +function functionResponseName(record: ToolResultRecord): string | undefined { + for (const part of record.parts) { + if (part.kind === "function_response" && part.name.length > 0) return part.name; + } + return undefined; +} + +export function buildTranscriptFlow(records: ReadonlyArray): FlowItem[] { + // records are TranscriptRecord; typed loosely to avoid a wide import union here. + const items: FlowItem[] = []; + let currentAssistant: AssistantFlowItem | null = null; + let pendingLatencyMs: number | undefined; + + for (const raw of records) { + const record = raw as { type: string } & Record; + + if (record.type === "user") { + currentAssistant = null; + items.push({ kind: "user", record: raw as UserRecord }); + continue; + } + + if (record.type === "assistant") { + const assistant = raw as AssistantRecord; + const tools: ToolStep[] = []; + assistant.parts.forEach((part, index) => { + if (part.kind === "function_call") { + tools.push({ + key: `${assistant.uuid}-${index}`, + name: part.name, + args: part.args, + command: shellCommand(part.name, part.args), + status: "running", + durationMs: undefined, + error: undefined, + display: undefined, + response: [], + }); + } + }); + const item: AssistantFlowItem = { + kind: "assistant", + record: assistant, + latencyMs: pendingLatencyMs, + tools, + }; + pendingLatencyMs = undefined; + currentAssistant = item; + items.push(item); + continue; + } + + if (record.type === "tool_result") { + const result = raw as ToolResultRecord; + const name = functionResponseName(result); + const step = pickStep(currentAssistant, name, (candidate) => candidate.display === undefined); + if (step) { + step.display = result.toolCall?.display; + step.response = result.parts.filter((part) => part.kind === "function_response"); + // A `tool_result` is written ONLY for a CompletedToolCall, whose status is + // always success | error | cancelled (qwen coreToolScheduler.ts:165 + + // checkAndNotifyCompletion gate :1583). Its presence is therefore terminal, + // so anything that is not cancel/error resolves to success — a step that has + // a result must never stay "running". + const status = result.toolCall?.status; + step.status = + status === "cancelled" ? "cancelled" : status === "error" ? "error" : "success"; + } + continue; + } + + if (record.type === "system") { + const telemetry = readTelemetry(raw as SystemRecord); + if (!telemetry) { + // compaction / slash_command / at_command — a real timeline divider. + currentAssistant = null; + items.push({ kind: "system", record: raw as SystemRecord }); + continue; + } + // Match by suffix, not the full `.` brand prefix, so a CLI rebrand + // (e.g. "qwen-code." → "ru-code.") keeps working without a code change. + if (telemetry.name.endsWith(".api_response")) { + // Latency of the response that is about to be (or was just) emitted. + pendingLatencyMs = telemetry.durationMs; + if (currentAssistant && currentAssistant.latencyMs === undefined) { + currentAssistant.latencyMs = telemetry.durationMs; + } + } else if (telemetry.name.endsWith(".tool_call")) { + const step = pickStep( + currentAssistant, + telemetry.functionName, + (candidate) => candidate.durationMs === undefined, + ); + if (step) { + step.durationMs = telemetry.durationMs; + if (step.command === undefined) { + step.command = shellCommand(step.name, telemetry.functionArgs); + } + if (telemetry.status === "error") { + const cancelled = (telemetry.error ?? "").toLowerCase().includes("cancel"); + step.status = cancelled ? "cancelled" : "error"; + step.error = telemetry.error; + } else if (telemetry.status === "success" && step.status === "running") { + step.status = "success"; + } + } + } + // any other telemetry event is intentionally ignored + continue; + } + } + + return items; +} + +/** First tool in the current assistant turn matching `name` (if given) for which + * `open` still holds — so call/result/telemetry land on the same step. */ +function pickStep( + assistant: AssistantFlowItem | null, + name: string | undefined, + open: (step: ToolStep) => boolean, +): ToolStep | undefined { + if (!assistant) return undefined; + const byName = name + ? assistant.tools.find((step) => step.name === name && open(step)) + : undefined; + return byName ?? assistant.tools.find(open); +} diff --git a/apps/web/src/ru-fork/advanced-chat/transcriptTypes.ts b/apps/web/src/ru-fork/advanced-chat/transcriptTypes.ts new file mode 100644 index 00000000000..233688bdf58 --- /dev/null +++ b/apps/web/src/ru-fork/advanced-chat/transcriptTypes.ts @@ -0,0 +1,7 @@ +// ru-fork: advanced chat mode — per-type narrowings of the TranscriptRecord union. +import type { TranscriptRecord } from "@t3tools/contracts"; + +export type UserRecord = Extract; +export type AssistantRecord = Extract; +export type ToolResultRecord = Extract; +export type SystemRecord = Extract; diff --git a/apps/web/src/ru-fork/advanced-chat/transcriptVisuals.tsx b/apps/web/src/ru-fork/advanced-chat/transcriptVisuals.tsx new file mode 100644 index 00000000000..b272a4ff79f --- /dev/null +++ b/apps/web/src/ru-fork/advanced-chat/transcriptVisuals.tsx @@ -0,0 +1,82 @@ +// ru-fork: advanced chat mode — shared visual primitives for the transcript view. +// Theme-safe (semantic tokens only), nothing truncated: heavy content goes into +// a (collapsed) so everything is reachable without clamping. +import { CheckIcon, ChevronRightIcon, CopyIcon } from "lucide-react"; +import { type ReactNode, useState } from "react"; + +import { Collapsible, CollapsibleContent, CollapsibleTrigger } from "~/components/ui/collapsible"; +import { cn } from "~/lib/utils"; + +/** Collapsible section with a chevron + label and an optional right-aligned meta slot. */ +export function Disclosure({ + label, + meta, + defaultOpen = false, + tone = "muted", + children, +}: { + label: ReactNode; + meta?: ReactNode; + defaultOpen?: boolean; + tone?: "muted" | "default"; + children: ReactNode; +}) { + const [open, setOpen] = useState(defaultOpen); + return ( + + + + {label} + {meta ? ( + {meta} + ) : null} + + +
{children}
+
+
+ ); +} + +/** Monospace JSON / text block. Horizontal scroll only (never clips vertically). */ +export function JsonBlock({ value }: { value: unknown }) { + const text = (() => { + try { + return typeof value === "string" ? value : JSON.stringify(value, null, 2); + } catch { + return String(value); + } + })(); + return ( +
+      {text}
+    
+ ); +} + +/** Copy-to-clipboard icon button, matching the chat bubble affordance. */ +export function CopyButton({ text }: { text: string }) { + const [copied, setCopied] = useState(false); + return ( + + ); +} diff --git a/apps/web/src/ru-fork/advanced-chat/useTranscriptSubscription.ts b/apps/web/src/ru-fork/advanced-chat/useTranscriptSubscription.ts new file mode 100644 index 00000000000..a4fcfbfaad3 --- /dev/null +++ b/apps/web/src/ru-fork/advanced-chat/useTranscriptSubscription.ts @@ -0,0 +1,53 @@ +// ru-fork: advanced chat mode — subscribe to the server's qwen-transcript stream for +// a thread and keep the records in local state. Self-contained: reuses the exported +// environment-connection registry (no store.ts / service.ts changes). Re-attaches if +// the environment connection object is replaced (full reconnect). +import type { EnvironmentId, ThreadId, TranscriptRecord } from "@t3tools/contracts"; +import { useEffect, useState } from "react"; + +import { + readEnvironmentConnection, + subscribeEnvironmentConnections, +} from "~/environments/runtime/service"; + +export function useTranscriptSubscription( + environmentId: EnvironmentId | undefined, + threadId: ThreadId | undefined, + enabled: boolean, +): ReadonlyArray { + const [records, setRecords] = useState>([]); + + useEffect(() => { + if (!enabled || !environmentId || !threadId) { + setRecords([]); + return; + } + setRecords([]); + + let unsubscribe = () => {}; + let attachedTo: unknown = null; + + const attach = () => { + const connection = readEnvironmentConnection(environmentId); + if (!connection || connection === attachedTo) return; + unsubscribe(); + attachedTo = connection; + unsubscribe = connection.client.orchestration.subscribeTranscript({ threadId }, (item) => { + if (item.kind === "snapshot") { + setRecords(item.records); + } else { + setRecords((previous) => [...previous, ...item.records]); + } + }); + }; + + attach(); + const unsubscribeConnections = subscribeEnvironmentConnections(attach); + return () => { + unsubscribe(); + unsubscribeConnections(); + }; + }, [environmentId, threadId, enabled]); + + return records; +} diff --git a/apps/web/src/ru-fork/config.ts b/apps/web/src/ru-fork/config.ts index 61b18ac8b27..620ac1628d9 100644 --- a/apps/web/src/ru-fork/config.ts +++ b/apps/web/src/ru-fork/config.ts @@ -12,3 +12,15 @@ export const DISABLE_AUTO_APPROVE = true; // HIDE_EXTRA_FEATURES so we can re-enable model selection while keeping // other multi-provider surfaces hidden in the single-provider build. export const SHOW_MODEL_SELECTOR = true; + +// ru-fork: advanced chat mode diff renderer. true = rich @pierre/diffs FileDiff +// (clean gutter, no patch headers); false = raw Shiki unified-patch fallback. +// Flip to false to fall back instantly if the rich renderer misbehaves. +export const ADVANCED_CHAT_RICH_DIFF = true; + +// ru-fork: advanced chat mode — preview a pending file operation before it is +// approved/applied. true = render an un-applied `edit` as a synthesized +// old→new diff (jsdiff createTwoFilesPatch) and a pending `write_file` as a +// syntax-highlighted code block; false = just show the raw call arguments. +// Orthogonal to ADVANCED_CHAT_RICH_DIFF, which only styles a diff once shown. +export const ADVANCED_CHAT_PENDING_PREVIEW = true; diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 289d9182395..91c260c6a89 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -16,6 +16,8 @@ export * from "./ru-fork/skills.ts"; export * from "./ru-fork/subagents.ts"; export * from "./ru-fork/terminalErrors.ts"; export * from "./ru-fork/mcp.ts"; +// ru-fork: advanced chat mode — normalized qwen transcript contract. +export * from "./ru-fork/transcript.ts"; export * from "./settings.ts"; export * from "./git.ts"; export * from "./vcs.ts"; diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index e5b8b192be4..7c93dbf19fc 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -93,6 +93,12 @@ import { } from "./ru-fork/mcp.ts"; import { ProjectId } from "./baseSchemas.ts"; import { ServerSettings, ServerSettingsError, ServerSettingsPatch } from "./settings.ts"; +// ru-fork: advanced chat mode — transcript subscription. +import { + TRANSCRIPT_WS_METHOD, + TranscriptRpcSchemas, + TranscriptSubscribeError, +} from "./ru-fork/transcript.ts"; import { SourceControlCloneRepositoryInput, SourceControlCloneRepositoryResult, @@ -474,6 +480,14 @@ export const WsOrchestrationSubscribeThreadRpc = Rpc.make( }, ); +// ru-fork: advanced chat mode — stream normalized qwen transcript records. +export const WsOrchestrationSubscribeTranscriptRpc = Rpc.make(TRANSCRIPT_WS_METHOD, { + payload: TranscriptRpcSchemas.subscribeTranscript.input, + success: TranscriptRpcSchemas.subscribeTranscript.output, + error: TranscriptSubscribeError, + stream: true, +}); + export const WsSubscribeTerminalEventsRpc = Rpc.make(WS_METHODS.subscribeTerminalEvents, { payload: Schema.Struct({}), success: TerminalEvent, @@ -594,4 +608,6 @@ export const WsRpcGroup = RpcGroup.make( WsMcpRecheckRpc, WsSubscribeMcpProjectionRpc, WsSubscribeMcpRuntimeRpc, + // ru-fork: advanced chat mode. + WsOrchestrationSubscribeTranscriptRpc, ); diff --git a/packages/contracts/src/ru-fork/transcript.ts b/packages/contracts/src/ru-fork/transcript.ts new file mode 100644 index 00000000000..8eb3e92d5a0 --- /dev/null +++ b/packages/contracts/src/ru-fork/transcript.ts @@ -0,0 +1,179 @@ +// ru-fork: contract for "advanced chat mode" — a normalized view of qwen's +// on-disk JSONL conversation transcript (read-only). The server's +// QwenTranscriptService parses qwen `ChatRecord` lines into these shapes and +// streams them; the web client renders them. Kept under ru-fork/ so upstream +// re-syncs never collide. See ru-fork-instrumental/advanced-chat/PLAN.md. +import * as Schema from "effect/Schema"; + +import { ThreadId } from "../baseSchemas.ts"; + +/** WS method name. Defined here (not in orchestration.ts) to avoid touching upstream. */ +export const TRANSCRIPT_WS_METHOD = "orchestration.subscribeTranscript" as const; + +/** + * One normalized part of a message. Maps from a @google/genai `Part` — we read + * the raw JSON structurally (qwen geminiChat.ts discriminates the same way). + */ +export const TranscriptPart = Schema.Union([ + Schema.Struct({ kind: Schema.Literal("text"), text: Schema.String }), + Schema.Struct({ kind: Schema.Literal("thought"), text: Schema.String }), + Schema.Struct({ + kind: Schema.Literal("function_call"), + name: Schema.String, + args: Schema.Unknown, + }), + Schema.Struct({ + kind: Schema.Literal("function_response"), + name: Schema.String, + response: Schema.Unknown, + }), + Schema.Struct({ kind: Schema.Literal("inline_data"), mimeType: Schema.String }), + Schema.Struct({ kind: Schema.Literal("unknown"), raw: Schema.Unknown }), +]); +export type TranscriptPart = typeof TranscriptPart.Type; + +/** qwen CoreToolScheduler `Status`. */ +export const TranscriptToolStatus = Schema.Literals([ + "validating", + "scheduled", + "error", + "success", + "executing", + "cancelled", + "awaiting_approval", +]); +export type TranscriptToolStatus = typeof TranscriptToolStatus.Type; + +/** Normalized `ToolResultDisplay` union (qwen tools.ts:532). */ +export const TranscriptToolDisplay = Schema.Union([ + Schema.Struct({ kind: Schema.Literal("text"), text: Schema.String }), + Schema.Struct({ + kind: Schema.Literal("file_diff"), + fileName: Schema.String, + fileDiff: Schema.String, + originalContent: Schema.NullOr(Schema.String), + newContent: Schema.String, + diffStat: Schema.optional(Schema.Unknown), + }), + Schema.Struct({ + kind: Schema.Literal("todo_list"), + todos: Schema.Array( + Schema.Struct({ + id: Schema.String, + content: Schema.String, + status: Schema.Literals(["pending", "in_progress", "completed"]), + }), + ), + }), + Schema.Struct({ + kind: Schema.Literal("plan_summary"), + message: Schema.String, + plan: Schema.String, + rejected: Schema.optional(Schema.Boolean), + }), + Schema.Struct({ + kind: Schema.Literal("task_execution"), + subagentName: Schema.String, + status: Schema.String, + taskDescription: Schema.String, + result: Schema.optional(Schema.String), + toolCalls: Schema.optional(Schema.Array(Schema.Unknown)), + }), + Schema.Struct({ + kind: Schema.Literal("mcp_progress"), + progress: Schema.Number, + total: Schema.optional(Schema.Number), + message: Schema.optional(Schema.String), + }), + Schema.Struct({ kind: Schema.Literal("ansi"), raw: Schema.Unknown }), + Schema.Struct({ kind: Schema.Literal("unknown"), raw: Schema.Unknown }), +]); +export type TranscriptToolDisplay = typeof TranscriptToolDisplay.Type; + +export const TranscriptToolCall = Schema.Struct({ + callId: Schema.optional(Schema.String), + status: Schema.optional(TranscriptToolStatus), + display: Schema.optional(TranscriptToolDisplay), +}); +export type TranscriptToolCall = typeof TranscriptToolCall.Type; + +export const TranscriptUsage = Schema.Struct({ + promptTokens: Schema.optional(Schema.Number), + cachedTokens: Schema.optional(Schema.Number), + outputTokens: Schema.optional(Schema.Number), + totalTokens: Schema.optional(Schema.Number), +}); +export type TranscriptUsage = typeof TranscriptUsage.Type; + +const TranscriptBaseFields = { + uuid: Schema.String, + parentUuid: Schema.NullOr(Schema.String), + sessionId: Schema.String, + timestamp: Schema.String, + cwd: Schema.String, + gitBranch: Schema.optional(Schema.String), +} as const; + +export const TranscriptSystemSubtype = Schema.Literals([ + "chat_compression", + "slash_command", + "ui_telemetry", + "at_command", +]); +export type TranscriptSystemSubtype = typeof TranscriptSystemSubtype.Type; + +export const TranscriptRecord = Schema.Union([ + Schema.Struct({ + ...TranscriptBaseFields, + type: Schema.Literal("user"), + parts: Schema.Array(TranscriptPart), + }), + Schema.Struct({ + ...TranscriptBaseFields, + type: Schema.Literal("assistant"), + parts: Schema.Array(TranscriptPart), + model: Schema.optional(Schema.String), + contextWindowSize: Schema.optional(Schema.Number), + usage: Schema.optional(TranscriptUsage), + }), + Schema.Struct({ + ...TranscriptBaseFields, + type: Schema.Literal("tool_result"), + parts: Schema.Array(TranscriptPart), + toolCall: Schema.optional(TranscriptToolCall), + }), + Schema.Struct({ + ...TranscriptBaseFields, + type: Schema.Literal("system"), + subtype: Schema.optional(TranscriptSystemSubtype), + payload: Schema.optional(Schema.Unknown), + }), +]); +export type TranscriptRecord = typeof TranscriptRecord.Type; + +export const TranscriptStreamItem = Schema.Union([ + Schema.Struct({ + kind: Schema.Literal("snapshot"), + records: Schema.Array(TranscriptRecord), + }), + Schema.Struct({ + kind: Schema.Literal("append"), + records: Schema.Array(TranscriptRecord), + }), +]); +export type TranscriptStreamItem = typeof TranscriptStreamItem.Type; + +export class TranscriptSubscribeError extends Schema.TaggedErrorClass()( + "TranscriptSubscribeError", + { + message: Schema.String, + threadId: Schema.String, + }, +) {} + +export const TranscriptRpcSchemas = { + subscribeTranscript: { + input: Schema.Struct({ threadId: ThreadId }), + output: TranscriptStreamItem, + }, +} as const;