deepseek-harness/apps/web/tests/scaffold.ts

// Shared scaffold for the keyless browser e2e lane (Agent Note:
// .agents/notes/implemented/testing/2026-07-24-web-gui-browser-e2e-lane.md).
// Boots the REAL web composition — the dsh-base and dsh-web-app bundle
// patches over the empty profile root through the vendored Loader (the same
// layer stack the profile boot composes), patched the
// snapshot way — so a real chromium exercises the real HTTP uplink/WebSocket
// downlink, api-gateway, agent loop, tools, and persistence. Modes ride $DSH_SNAPSHOT:
// replay (default, keyless: normally disables the direct DeepSeek rows and
// inserts dsh-llm-replay in providers mode), record (real adapter + key,
// harvests fixtures from live session memory), refresh (keyless replay that
// rewrites goldens). A first-run option keeps the real adapter mounted while
// masking its credential, without making a model call.
//
// Composition divergences from `dsh web`, all deliberate, all via include
// patches after the shipped bundle layers, over the SAME tree (never a
// second yml): temp persistenceRoot; host-level skill roots confined to the
// temp workspace while project skill discovery remains real; agent-instructions
// disabled (recorded fixtures must not embed this repo's AGENTS.md);
// session-title-llm disabled (its fire-and-forget title call would race the
// loop for the session's replay cursor); webserver pinned to port 0 with the
// built dist; ordinary keyless modes disable both direct adapters and fill the open
// llm seam post-boot with installLlmReplay on the settled root ctx
// (the plugin-row path discards the ReplayHandle; the direct install keeps
// assertConsumed for the teardown fixture-consumption check).
import { existsSync, readFileSync } from 'node:fs'
import { createHash } from 'node:crypto'
import { mkdir, mkdtemp, readFile, readdir, realpath, rm, symlink, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { basename, dirname, join, resolve } from 'node:path'
import { pathToFileURL } from 'node:url'
import type { Page } from 'playwright'
import { expect } from 'vitest'
import { Context } from '@deepseek-ai/cordis'
import { DSH_LAUNCH_ENVIRONMENT_KEY, type LaunchEnvironmentSnapshot } from '@deepseek-ai/dsh-launch-environment'
import Loader from '@deepseek-ai/cordis-plugin-loader'
import Include, { type PatchOptions } from '@deepseek-ai/cordis-plugin-include'
import Group from '@deepseek-ai/cordis-plugin-group'
import {
  captureExpectedWorkspaceSnapshot,
  captureWorkspaceSnapshot,
  assertSessionFixtureVersion,
  formatSystemPromptSnapshot,
  formatToolSchemasSnapshot,
  normalizedSystemPrompts,
  normalizedToolSchemas,
  parseSnapshotManifest,
  redactSessionSnapshotIds,
  normalizeSessionSnapshots,
  parseSessionFixtureName,
  scrubModelRequestBulk,
  scrubSessionSnapshot,
  sessionFixtureFiles,
  sessionFixtureName,
  stabilizeFixtureMessageIds,
  stabilizeRefreshLog,
  writesCurrentSessionFixtures,
  type NormalizeContext,
} from '@deepseek-ai/dsh-session-snapshot'
import type { Profile, ProfileContext, ProfileResolutionMode } from '@deepseek-ai/dsh-app-boot'
import { dshHomePath } from '@deepseek-ai/dsh-home-paths'
import { LlmAdapter } from '@deepseek-ai/dsh-llm'
import type {
  LlmModelInfo, LlmProviderInfo, LlmResolvedModelInfo, RetryPolicyConfig, StreamChunk,
} from '@deepseek-ai/dsh-llm'
import type { ReplayHandle, ReplayProviderConfig } from '@deepseek-ai/dsh-llm-replay'
import {
  installLlmReplay,
  parseSessionLog,
  prepareSessionSnapshotFixtureForComparison,
} from '@deepseek-ai/dsh-llm-replay'
import type { SessionFormatEvent } from '@deepseek-ai/dsh-session-format'
import { sessionFormatCatalog } from '@deepseek-ai/dsh-session-format-catalog'
import {
  SESSION_FORMAT_VERSION,
  SessionId,
  type Session,
  type SessionEvent,
  type SessionHeader,
} from '@deepseek-ai/dsh-session'
import JsonlSessionPersistence from '@deepseek-ai/dsh-session-persistence-jsonl'
// Empty type imports carry the webServer/agents/sessionPersistence Context merges.
import type {} from '@deepseek-ai/dsh-host-webserver'
import type {} from '@deepseek-ai/dsh-agent'
import { provideCmdline } from '@deepseek-ai/dsh-cmdline'
import { REPO_ROOT, requireBuilt, requireDist } from './support.ts'

type AppBoot = typeof import('@deepseek-ai/dsh-app-boot')
let builtAppBoot: AppBoot | undefined

/**
 * The launcher's own module, as built: the manager and HMR plugins the profile
 * loads reload the tree through this copy's registry of the root Include, so
 * the scaffold mounts through the same copy rather than the source import.
 * Resolved on the first launch, which needs the build anyway, so the fixture
 * helpers this module also exports load without one.
 */
function appBoot(): AppBoot {
  builtAppBoot ??= requireBuilt('@deepseek-ai/dsh-app-boot') as AppBoot
  return builtAppBoot
}

// Host-side web e2e cannot import a browser package: doing so would pull that
// package's complete TS project into this graph. Mirrored from
// packages/client/ui-settings-models/src/onboarding-copy.ts; drift makes the
// default pre-acknowledgement stop suppressing the notice and fails loudly.
// import {
//   WELCOME_NOTICE_ACK_FIELD, WELCOME_NOTICE_SETTINGS_NAMESPACE,
//   WELCOME_NOTICE_VERSION, WELCOME_NOTICE_COPY,
// } from '@deepseek-ai/dsh-client-ui-settings-models'
export const WELCOME_NOTICE_SETTINGS_NAMESPACE = 'ui-onboarding'
export const WELCOME_NOTICE_ACK_FIELD = 'welcomeNoticeVersion'
export const WELCOME_NOTICE_VERSION = '2026-08-13.1'
export const WELCOME_NOTICE_COPY = {
  zh: {
    title: '内测声明',
    body: 'DeepSeek Harness 目前的 0.1 版本仍处在面向 Harness 开发者进行测试的阶段，还有许多地方需要持续改进和打磨，希望听取广大开发者的反馈建议。预计 DeepSeek Harness 的核心插件以及基础 API 都会在接下来的一段时间内快速迭代、持续演化。\n\n我们期待与全球开发者一起，在开源、开放、可复用、可组合的基础设施之上，共同探索智能上限。欢迎全球 Harness 开发者加入 DSH 插件生态。',
    continueLabel: '继续',
  },
} as const

/** Snapshot mode for the lane, from $DSH_SNAPSHOT (same vocabulary as the other snapshot suites). */
export type WebSnapshotMode = 'replay' | 'record' | 'refresh'

/**
 * Resolve and validate the lane's snapshot mode.
 * @returns the active mode; unset/empty selects replay.
 */
export function webSnapshotMode(): WebSnapshotMode {
  const value = process.env.DSH_SNAPSHOT
  if (value === undefined || value === '' || value === 'replay') return 'replay'
  if (value === 'record' || value === 'refresh') return value
  throw new Error(`DSH_SNAPSHOT must be replay, record, or refresh; got ${JSON.stringify(value)}`)
}

/**
 * Compare a session-driven Web scenario's complete workspace with its committed independent expected state.
 * @param scenarioDir - Absolute recorded-session scenario directory.
 * @param workspaceRoot - Absolute cwd used by the controlled session.
 */
export async function assertFinalWorkspaceSnapshot(scenarioDir: string, workspaceRoot: string): Promise<void> {
  const manifestPath = join(scenarioDir, 'snapshot.yml')
  const manifest = parseSnapshotManifest(await readFile(manifestPath, 'utf8'), manifestPath)
  expect(manifest.workspace?.final, `${manifest.scenario ?? scenarioDir}: mutating Web scenario declares workspace.final`)
    .toBe(true)
  const actual = await captureWorkspaceSnapshot(workspaceRoot)
  const expected = await captureExpectedWorkspaceSnapshot(join(scenarioDir, 'workspace.expected'))
  expect(actual, `${manifest.scenario ?? scenarioDir}: complete final workspace`).toEqual(expected)
}

async function ownsReplayFixture(replayFixture: string | undefined): Promise<boolean> {
  if (replayFixture === undefined) return false
  const fixture = parseSessionFixtureName(basename(replayFixture))
  if (fixture === undefined || fixture.index !== 0) return false
  const manifestPath = join(dirname(replayFixture), 'snapshot.yml')
  if (!existsSync(manifestPath)) return false
  const manifest = parseSnapshotManifest(await readFile(manifestPath, 'utf8'), manifestPath)
  return manifest.session === undefined
}

/**
 * Resolve one requested fixture role to its highest committed generation.
 * @param path - any generation path for the requested parent or child role.
 * @param allowAbsent - Keep an absent canonical path only for an override-only replay script.
 * @returns the highest canonical sibling generation, or the input for non-Session files.
 */
export async function selectedSessionFixture(path: string, allowAbsent = false): Promise<string> {
  const requested = parseSessionFixtureName(basename(path))
  if (requested === undefined) return path
  const entries = await readdir(dirname(path))
  if (allowAbsent && !entries.some(name => parseSessionFixtureName(name) !== undefined)) return path
  const selected = sessionFixtureFiles(entries)
    .find(candidate => candidate.index === requested.index)
  if (selected === undefined) throw new Error(`${path}: missing Session fixture role ${requested.index}`)
  const resolved = join(dirname(path), selected.name)
  assertSessionFixtureVersion(selected.name, await readFile(resolved, 'utf8'))
  return resolved
}

/**
 * Return the current-writer target without replacing the requested older fixture.
 * @param path - any canonical fixture generation for one role.
 * @param version - generation emitted by the current writer.
 * @returns the canonical sibling path for that role and generation.
 */
export function recordedSessionFixturePath(path: string, version: number): string {
  const fixture = parseSessionFixtureName(basename(path))
  if (fixture === undefined) throw new Error(`record harvest: invalid Session fixture path ${path}`)
  return join(dirname(path), sessionFixtureName(fixture.index, version))
}

/** The shipped composition under test: the dsh-base and dsh-web-app bundle patches over the empty profile root. */
const BASE_PATCH_PATH = join(REPO_ROOT, 'packages/bundle/base/cordis.patch.yml')
const WEB_PATCH_PATH = join(REPO_ROOT, 'packages/bundle/web-app/cordis.patch.yml')
/** The installation anchor whose dependency surface the profile module fallback mirrors. */
const INSTALL_ANCHOR = join(REPO_ROOT, 'apps/cli/package.json')

// Replay publishes the provider catalog the gateway routes to (providers
// mode, never catch-all: with both direct adapters disabled no adapter exists, so a
// catch-all would leave resolveModelInfo unroutable and compaction-basic's
// post-step pressure check would warn every step). The published
// contextWindow keeps that pressure path provably inert for small fixtures.
const REPLAY_PROVIDERS = [{
  id: 'deepseek-official',
  name: 'DeepSeek',
  models: [
    { id: 'deepseek-v4-flash', name: 'DeepSeek-V4-Flash', contextWindow: 128_000 },
    {
      id: 'deepseek-v4-flash-vision-exp',
      name: 'DeepSeek-V4-Flash-Vision-Exp',
      contextWindow: 1_000_000,
      inputModalities: ['text', 'image'] as const,
      defaultMaxTokens: 256_000,
      reasoningEfforts: ['off', 'low', 'high', 'max'],
      defaultReasoningEffort: 'high',
    },
  ],
}]

/**
 * The routes a shipped composition always has, with no ability to stream.
 * A fixture-less keyless scenario issues no model calls, but its tree must
 * still answer `listProviders()` — surfaces legitimately gate on whether any
 * adapter serves a session's route, and an empty registry is a test artifact,
 * not a product state.
 */
class RouteOnlyAdapter extends LlmAdapter {
  constructor(private readonly providers: typeof REPLAY_PROVIDERS) {
    super()
  }

  override providerInfo(provider: string): LlmProviderInfo {
    return { id: provider, name: this.providers.find(entry => entry.id === provider)?.name ?? provider }
  }

  override listModels(provider: string): Promise<readonly LlmModelInfo[]> {
    return Promise.resolve((this.providers.find(entry => entry.id === provider)?.models ?? [])
      .map(model => ({ provider, id: model.id, name: model.name })))
  }

  override resolveModel(provider: string, model: string): Promise<LlmResolvedModelInfo> {
    const listed = this.providers.find(entry => entry.id === provider)?.models
      .find(entry => entry.id === model)
    return Promise.resolve({
      provider,
      id: model,
      name: listed?.name ?? model,
      ...listed?.contextWindow === undefined ? {} : { contextWindow: listed.contextWindow },
    })
  }

  override async *stream(): AsyncIterable<StreamChunk> {
    throw new Error(
      'web e2e scaffold: a model call was issued by a scenario that declared no replay fixture'
      + ' — pass replayFixture, or keep the scenario free of model calls',
    )
  }
}

function replayProviders(contextWindow: number | undefined, messages: boolean): typeof REPLAY_PROVIDERS {
  return REPLAY_PROVIDERS.map(provider => ({
    ...provider,
    id: messages ? 'deepseek-messages' : provider.id,
    models: provider.models.map(model => ({
      ...model,
      ...contextWindow === undefined ? {} : { contextWindow },
    })),
  }))
}

/** A booted web scaffold: real composition, mode-selected model backend, temp world. */
export interface WebScaffold {
  /** The active snapshot mode this scaffold booted under. */
  mode: WebSnapshotMode
  /** Browser-facing origin for the bound test server. */
  baseUrl: string
  /** Process-token URL that establishes this scaffold's browser session. */
  authenticatedUrl: string
  /** Settled root context (the in-process readiness barrier; headless event subscription is its sanctioned use). */
  ctx: Context
  /** Temp project directory sessions run in (shell/fs tool cwd). */
  workspaceCwd: string
  /** Temp persistence root (seeded sessions land here through the real API). */
  persistenceRoot: string
  /** Isolated harness home the settings/credentials rows write ($DSH_HOME double). */
  harnessHome: string
  /** Send a browser-equivalent Host request with this scaffold's authenticated cookie. */
  hostFetch(path: string, init?: RequestInit): Promise<Response>
  /** Await a settled turn end: in-process turn/end, then the agent's idle flip (which follows the persistence flush). */
  whenTurnSettled(timeoutMs?: number): Promise<SessionId>
  /**
   * Tear everything down; asserts the replay fixture was fully consumed first
   * (replay/refresh), unless booted with replayProvidersOnly (whose fixture
   * is validated call-free at boot).
   */
  close(): Promise<void>
}

/** Options for {@link launchWebScaffold}. */
export interface LaunchOptions {
  /** Profile resolver backend used by this test Host; defaults to runtime coverage. */
  profileResolutionMode?: Extract<ProfileResolutionMode, 'dual' | 'runtime'>
  /** Enable the real Open In rows with deterministic launch-environment facts. */
  openInAppEnvironment?: LaunchEnvironmentSnapshot
  /** Compare the replayed root session with `replayFixture`; defaults on for a manifest-owned canonical recording. */
  compareReplaySession?: boolean
  /**
   * Optional product overlay applied after the shipped Web surface and before
   * the scaffold's hermetic test patches, matching the launcher's `--patch`
   * ordering.
   */
  extraOverlayPath?: string
  /**
   * Additional package manifests whose dependency closures supply experimental
   * profile layers named by {@link extraOverlayPath}.
   */
  extraInstallAnchors?: string[]
  /**
   * Manage the scaffold profile the way the launcher does: a `profileContext`
   * over the profile directory, whose manifest lists the shipped web bundles
   * and each package directory here as an installed dependency (`file:` in the
   * manifest, a symlink under the profile's `node_modules`); `enabled` also
   * lists a bundle in `dsh.profile.bundles`. The plugin manager mounts on such
   * a profile, and the root Include is mounted from the profile's own layers.
   * The base bundle's `hmr` row turns on with the profile context, so
   * configuration changes apply live; `hmr: false` disables that row through
   * an overlay, leaving changes for the next start.
   */
  profile?: {
    hmr?: boolean
    packages: { dir: string; enabled?: boolean }[]
  }
  /**
   * Replay fixture (session.jsonl) served by the inserted dsh-llm-replay row
   * in replay/refresh modes; ignored in record mode (the real adapter
   * answers). Omit for scenarios issuing no model calls — a stray stream then
   * fails loud with NO_ADAPTER (both direct adapters are disabled and no replay row
   * mounts). With {@link replayProvidersOnly}, the fixture must record no
   * model calls (its header alone mounts the catalog).
   */
  replayFixture?: string
  /** Explicit replay routes for scenarios exercising provider-dependent behavior; replay/refresh only. */
  replayProviders?: ReplayProviderConfig[]
  /**
   * Mount the replay provider catalog (the model directory the UI shows)
   * without consuming any recorded script: for scenarios that never call a
   * model but need the real provider/model labels rendered. Requires
   * {@link replayFixture} whose log records no model calls, and rejects
   * {@link replayOverride} and {@link replayChildFixtures}; the teardown
   * consumption check is skipped for this mode. `replayFixture` without this
   * flag keeps the consumption check.
   */
  replayProvidersOnly?: boolean
  /**
   * Recorded child logs assigned in child creation order. Each child owns its
   * own positional replay cursor across initial and continuation turns.
   */
  replayChildFixtures?: string[]
  /**
   * Optional replay.override.json sidecar (whole-script replacement or
   * `{ patches }` augmentation) for throw/hang scenarios not expressible as
   * recorded chunks; replay/refresh only.
   */
  replayOverride?: string
  /**
   * Retry policy registered on every replay provider route, for failure-
   * injection scenarios that must exhaust recovery quickly instead of walking
   * the shared normal default's five backed-off retries; replay/refresh only.
   */
  replayRetryPolicy?: RetryPolicyConfig
  /** Per-chunk replay pacing (ms) so the browser observes genuinely incremental SSE; replay/refresh only. */
  paceMs?: number
  /** Synthetic model capacity for UI scenarios whose seeded history must remain uncompacted. */
  replayContextWindow?: number
  /**
   * Tool presentation mode patched onto the shipped `tools` row (`code`
   * collapses the wire to run_code + the SDK prompt section). Omit for the
   * yml default. The PTC runtime row is always in the tree, so no extra
   * insertion is needed.
   */
  toolsMode?: 'native' | 'ptc' | 'both'
  /**
   * Insert the opt-in model-facing Cordis tool provider into the shipped tree.
   * Record and replay use the same tool surface, so captured request headers
   * remain reconstructable without making the tools a product default.
   */
  cordisTools?: boolean
  /**
   * Keep the shipped DeepSeek adapter mounted while masking the process
   * environment's DEEPSEEK_API_KEY for this scaffold lifetime. This is the
   * keyless first-run configuration lane; the default disables the adapter.
   */
  deepSeekMissingCredential?: boolean
  /** Record or replay a Messages scenario; older scenarios explicitly retain their recorded Chat Completions route. */
  deepSeekMessages?: boolean
  /** Leave the current welcome notice pending; ordinary scenarios pre-acknowledge it before browser boot. */
  welcomeNoticePending?: boolean
  /**
   * Patch the shipped DeepSeek search row to a deterministic endpoint and
   * credential reference. Browser search scenarios keep the real provider and
   * credentials seam while avoiding external search traffic and ambient keys.
   */
  deepSeekSearch?: {
    /** Anthropic-compatible base URL; the provider appends `/messages`. */
    baseURL: string
    /** Credential reference resolved by the shipped search provider. */
    apiKeyEnv: string
  }
  /**
   * Replace the roster row the scaffold pins by default (no configured roots,
   * default `standard` — the plugin's own shipped presets). Supply this only
   * to change WHICH presets a scenario sees beyond the shipped set — a
   * writable user root, a different default. The patch lands after the
   * default, so it wins.
   */
  agentPresets?: {
    /** Roots to discover after the plugin's shipped root, in precedence order. */
    roots: { path: string; trust: 'system' | 'user' }[]
    /** The preset a session that names none is composed from. */
    default: string
  }
  /**
   * Patch the telemetry exporter URL while preserving the shipped enabled
   * setting. A scenario-owned loopback collector contains all fixture uploads.
   */
  telemetryUrl?: string
  /** Mode when telemetryUrl is supplied; defaults to FEEDBACK_ONLY without enabling a disabled row. */
  telemetryMode?: 'FEEDBACK_ONLY'
  /** SDK batch cadence for a scenario-owned collector; omitted to retain the SDK default. */
  telemetryScheduledDelayMillis?: number
  /**
   * Browse through a trusted non-loopback hostname that the browser resolves
   * to loopback (for example `*.localhost`). The test server stays bound to
   * 127.0.0.1; a non-resolving authority fails before Host trust is exercised.
   */
  remoteAuthority?: string
  /** Reuse an existing harness home so a second Host can verify user settings across origins. */
  harnessHome?: string
}

/** Dispose the booted tree and remove both owned temp roots, reporting every independent cleanup failure. */
async function cleanupScaffoldWorld(ctx: Context, workspaceCwd: string, persistenceRoot: string): Promise<unknown[]> {
  const failures: unknown[] = []
  await Promise.resolve(ctx.fiber.dispose()).catch((error: unknown) => failures.push(error))
  await rm(workspaceCwd, { recursive: true, force: true }).catch((error: unknown) => failures.push(error))
  await rm(persistenceRoot, { recursive: true, force: true }).catch((error: unknown) => failures.push(error))
  return failures
}

/**
 * Boot the real web composition under the current snapshot mode.
 * @param options - replay fixture selection and pacing.
 * @returns the running scaffold.
 */
export async function launchWebScaffold(options: LaunchOptions = {}): Promise<WebScaffold> {
  requireDist()
  const {
    auditStartupEntries, composeEntries, createProfileResolutionGeneration, healProfilesModuleFallback, initProfile,
    mountRootInclude, readProfileManifest, readProfilePatches, loadOverlayPatches, PluginPackages,
  } = appBoot()
  const mode = webSnapshotMode()
  const replayFixture = options.replayFixture === undefined
    ? undefined
    : await selectedSessionFixture(options.replayFixture, options.replayOverride !== undefined)
  const replayChildFixtures = options.replayChildFixtures === undefined
    ? undefined
    : await Promise.all(options.replayChildFixtures.map(path => selectedSessionFixture(path)))
  const compareReplaySession = options.compareReplaySession ?? await ownsReplayFixture(replayFixture)
  const browserHost = options.remoteAuthority ?? '127.0.0.1'
  if (mode === 'record') {
    // Both owning vitest configs (web unconditionally, snapshot in record
    // mode) load the repo-root .env before this file runs.
    if (process.env.DEEPSEEK_API_KEY === undefined || process.env.DEEPSEEK_API_KEY.length === 0) {
      throw new Error('web e2e record mode needs DEEPSEEK_API_KEY (env or repo-root .env)')
    }
  }
  if (mode === 'record' && options.deepSeekMissingCredential === true) {
    throw new Error('deepSeekMissingCredential is a keyless replay/refresh option')
  }
  const maskDeepSeekCredential = mode !== 'record' && options.deepSeekMissingCredential === true
  const messages = options.deepSeekMessages === true
  const originalDeepSeekCredential = process.env.DEEPSEEK_API_KEY
  let credentialEnvironmentRestored = false
  const restoreCredentialEnvironment = (): void => {
    if (credentialEnvironmentRestored || !maskDeepSeekCredential) return
    credentialEnvironmentRestored = true
    if (originalDeepSeekCredential === undefined) {
      Reflect.deleteProperty(process.env, 'DEEPSEEK_API_KEY')
    } else {
      process.env.DEEPSEEK_API_KEY = originalDeepSeekCredential
    }
  }
  const workspaceCwd = await realpath(await mkdtemp(join(tmpdir(), 'dsh-web-e2e-ws-')))
  // Isolated harness home: the settings/credentials rows resolve $DSH_HOME
  // paths at load, and an in-process boot must NEVER touch the developer's
  // real ~/.dsh document or credential file.
  const harnessHome = options.harnessHome ?? join(workspaceCwd, '.dsh-home')
  // Skill discovery is model-visible input, and its roots now resolve inside a
  // PRESET — a subtree this lane's include patches cannot reach, because the
  // roster mounts it directly per session rather than as a row of the booted
  // tree. The row's documented fallback is the environment, so pin that: the
  // whole scaffold lifetime, not just the boot, since presets mount when a
  // session is created. Without this a developer's real ~/.dsh/skills silently
  // enters replay requests and goldens while CI sees none. `DSH_HOME` follows
  // the resolved harness home so a scaffold sharing another's home — the
  // cross-port persistence scenario — pins the same roots the settings and
  // credentials rows were configured with.
  const skillRootEnvironment = {
    DSH_HOME: harnessHome,
    DSH_AGENTS_HOME: join(workspaceCwd, '.agents-home'),
    DSH_BUNDLED_SKILL_DIR: join(workspaceCwd, '.bundled-skills'),
  }
  const originalSkillRootEnvironment = Object.fromEntries(
    Object.keys(skillRootEnvironment).map(key => [key, process.env[key]]),
  )
  let skillRootEnvironmentRestored = false
  const restoreSkillRootEnvironment = (): void => {
    if (skillRootEnvironmentRestored) return
    skillRootEnvironmentRestored = true
    for (const [key, value] of Object.entries(originalSkillRootEnvironment)) {
      if (value === undefined) Reflect.deleteProperty(process.env, key)
      else process.env[key] = value
    }
  }
  Object.assign(process.env, skillRootEnvironment)
  let persistenceRoot: string
  try {
    persistenceRoot = await mkdtemp(join(tmpdir(), 'dsh-web-e2e-sessions-'))
  } catch (error) {
    const failures: unknown[] = [error]
    await rm(workspaceCwd, { recursive: true, force: true }).catch((cleanupError: unknown) => failures.push(cleanupError))
    restoreSkillRootEnvironment()
    if (failures.length > 1) throw new AggregateError(failures, 'web scaffold temp-root setup failed')
    throw error
  }
  if (maskDeepSeekCredential) Reflect.deleteProperty(process.env, 'DEEPSEEK_API_KEY')

  // The include patch set — the same layer stack the profile boot composes
  // (bundle patches in dsh.profile.bundles order), applied over the SAME empty root (a
  // patch id that stops matching a row fails the boot sweep loudly instead of
  // drifting).
  const basePatches = loadOverlayPatches('web e2e scaffold', BASE_PATCH_PATH)
  const surfacePatches = loadOverlayPatches('web e2e scaffold', WEB_PATCH_PATH)
  const extraOverlayPatches = options.extraOverlayPath === undefined
    ? []
    : loadOverlayPatches('web e2e scaffold', options.extraOverlayPath)
  const composedRows = composeEntries([basePatches, surfacePatches, extraOverlayPatches])
  const webRuntimeConfig = composedRows.find(row => row.id === 'web-runtime')?.config as {
    surfaceContext?: boolean
  } | undefined
  const surfaceContext = webRuntimeConfig?.surfaceContext !== false
  // The scaffold's own overrides, above every bundle layer like `--patch` overlays.
  const overlayPatches: PatchOptions[] = [
    // Without HMR the profile applies configuration changes at its next start.
    ...options.profile?.hmr === false ? [{ id: 'hmr', disabled: true }] : [],
    { id: 'session-log-deepseek', config: { enabled: false } },
    // The historical Messages fixture retains its recorded route during replay;
    // live configuration uses the shared DeepSeek route. Explicit overlays win.
    ...messages
      ? [{ id: 'agent-default-model', config: { provider: mode === 'record' || maskDeepSeekCredential ? 'deepseek-official' : 'deepseek-messages', model: maskDeepSeekCredential ? 'deepseek-flash' : 'deepseek-v4-flash' } }]
      : mode === 'record' || options.deepSeekMissingCredential === true
        ? []
        : [{ id: 'agent-default-model', config: { provider: 'deepseek-official', model: 'deepseek-v4-flash' } }],
    ...extraOverlayPatches,
    // The roster's shipped presets are the plugin's own, bundled inside
    // `dsh-agent-presets` and prepended by it. Pin only the machine-local
    // root away: a developer's own `~/.dsh/.agent-presets` must not be able
    // to change a golden.
    {
      id: 'agent-presets',
      config: {
        default: 'standard',
        includeUserRoot: false,
      },
    },
    { id: 'session-persistence-jsonl', config: { root: persistenceRoot } },
    // Content search is enabled here although the shipped bundles default it
    // off (`openAt: never`, pinned by apps/cli/tests/lazy-search-startup):
    // the seeded-session scenarios navigate by content search, and these e2e
    // runs are the assembled coverage for the opt-in search path.
    { id: 'session-query-sqlite', config: { path: ':memory:', openAt: 'first-search' } },
    // storage-json's yml root is anchored to the real $DSH_HOME; pin the row
    // to an absolute temp root (removed with the workspace at close) so tests
    // never write the user's harness home.
    { id: 'storage-json', config: { root: join(workspaceCwd, '.dsh-storages') } },
    // Skill discovery is model-visible input. Pin every host-level root inside
    // the owned temp world so ~/.dsh, ~/.agents, and a bundled-root env setting
    // cannot change replay requests or conversation goldens. Project roots stay
    // enabled against the same empty temp workspace, preserving the real seam.
    {
      id: 'skill-filesystem',
      config: {
        dshHome: join(workspaceCwd, '.dsh-home'),
        agentsHome: join(workspaceCwd, '.agents-home'),
        bundledSkillDir: join(workspaceCwd, '.bundled-skills'),
        watch: false,
      },
    },
    // fs/bash cwd default to process.cwd(); the gateway injects the same
    // value into session.cwd — chdir below anchors all three to the temp
    // workspace, keeping the composition untouched.
    { id: 'agent-instructions', disabled: true },
    { id: 'session-title-llm', disabled: true },
    // Fixture sessions must never leave the process: the shipped row defaults
    // to the production OTLP endpoint (or whatever DSH_TELEMETRY_OTLP_URL
    // names in the ambient environment). A scenario with a local collector
    // preserves the shipped disabled setting instead of overriding it.
    options.telemetryUrl === undefined
      ? { id: 'session-telemetry-otel', disabled: true }
      : {
        id: 'session-telemetry-otel',
        config: {
          mode: options.telemetryMode ?? 'FEEDBACK_ONLY',
          exporter: { url: options.telemetryUrl },
          ...(options.telemetryScheduledDelayMillis === undefined ? {} : {
            processor: { scheduledDelayMillis: options.telemetryScheduledDelayMillis },
          }),
          shutdownTimeoutMillis: 1_000,
        },
      },
    // Use an ephemeral port while preserving the shipped compression policy;
    // a patch replaces the row's complete config.
    {
      id: 'webserver',
      config: {
        host: '127.0.0.1', port: 0, compression: 'gzip',
        compressionLevel: 1, compressionThresholdBytes: 1024,
      },
    },
    // The bundle's web-runtime row resolves the same built dist under test
    // (apps/web IS @deepseek-ai/dsh-web-frontend); native browser opening and the
    // URL line are disabled because this scaffold owns its Playwright browser.
    // Preserve the composed surface-context choice because a patch replaces
    // the row's complete config.
    { id: 'web-runtime', config: { openBrowser: false, printUrl: false, surfaceContext } },
    ...options.remoteAuthority === undefined
      ? []
      : [{ id: 'connection', config: { trustedHosts: [options.remoteAuthority] } }],
    { id: 'settings', config: { dshHome: harnessHome } },
    { id: 'credentials', config: { dshHome: harnessHome } },
    // The shipped directory-picker row is the -auto chooser, which resolves
    // the interaction from the RUNNING host (display, SSH launch, bind). The
    // lane's goldens are interaction-specific (workspace-management drives
    // the in-app browse dialog), so pin -browse deterministically on every
    // host: patch `name` is an assertion, not an override, hence the
    // disable+insert pair.
    { id: 'directory-picker', disabled: true },
    { insert: [
      { id: 'directory-picker-browse', name: '@deepseek-ai/dsh-host-directory-picker-browse' },
      { id: 'ui-directory-picker-browse', name: '@deepseek-ai/dsh-client-ui-directory-picker-browse' },
    ] },
    // Ordinary scenarios exclude host-dependent application discovery. The
    // Open In scenario supplies launch facts that suppress every native probe.
    { id: 'open-in-app', disabled: options.openInAppEnvironment === undefined },
    { id: 'ui-open-in-app', disabled: options.openInAppEnvironment === undefined },
    ...options.agentPresets === undefined
      ? []
      // Never the derived harness-home root: a developer's own presets must not
      // be able to change a golden, whatever roots a scenario asks for.
      : [{ id: 'agent-presets', config: { ...options.agentPresets, includeUserRoot: false } }],
    ...options.toolsMode === undefined ? [] : [{ id: 'tools', config: { mode: options.toolsMode } }],
    // The shipped Web bundle already owns both runners and the Cordis UI. This
    // scenario adds only the model-facing tools that exercise those services.
    ...options.cordisTools === true
      ? [{ insert: [
        { id: 'tool-cordis', name: '@deepseek-ai/dsh-tool-cordis' },
      ] }]
      : [],
    ...options.deepSeekSearch === undefined
      ? []
      : [{
        id: 'web-search-deepseek',
        config: {
          apiKeyEnv: options.deepSeekSearch.apiKeyEnv,
          baseURL: options.deepSeekSearch.baseURL,
        },
      }],
    ...maskDeepSeekCredential && !messages ? [] : [
      { id: 'llm-deepseek', disabled: mode !== 'record' && !maskDeepSeekCredential,
        config: messages ? {} : { protocol: 'chat-completions' } },
    ],
  ]
  const patches: PatchOptions[] = [...basePatches, ...surfacePatches, ...overlayPatches]

  // Sessions inherit the gateway's process.cwd() default; run the boot from
  // the temp workspace so tool cwd, session cwd, and fixtures agree.
  const originalCwd = process.cwd()
  const ctx = new Context()
  if (options.openInAppEnvironment !== undefined) ctx.provide(DSH_LAUNCH_ENVIRONMENT_KEY, options.openInAppEnvironment)
  const observedSessions = new Map<SessionId, Session>()
  const stopObservingSessions = ctx.on('session/created', (session) => {
    observedSessions.set(session.id, session)
  })
  let port = 0
  let baseUrl = ''
  let authenticatedUrl = ''
  let cookieHeader = ''
  let replayHandle: ReplayHandle | undefined
  try {
    process.chdir(workspaceCwd)
    const profileDir = join(harnessHome, 'profiles', 'scaffold')
    const extraLayers: Profile['layers'] = await Promise.all((options.extraInstallAnchors ?? []).map(async (anchor) => {
      const manifest = JSON.parse(await readFile(anchor, 'utf8')) as { name?: unknown }
      if (typeof manifest.name !== 'string' || manifest.name === '') {
        throw new Error(`web scaffold extra install anchor has no package name: ${anchor}`)
      }
      const packageDir = dirname(anchor)
      // A real profile already has each bundle installed by `dsh plugin add`.
      // Reproduce that link so a private bundle can import its own plugin.
      const installedLink = join(profileDir, 'node_modules', manifest.name)
      await mkdir(dirname(installedLink), { recursive: true })
      await symlink(packageDir, installedLink, 'junction')
      return {
        packageName: manifest.name,
        packageDir,
        patchPath: join(packageDir, 'cordis.patch.yml'),
        patches: [],
      }
    }))
    const profile: Profile = {
      name: 'scaffold',
      dir: profileDir,
      layers: extraLayers,
      patchPath: join(profileDir, 'cordis.patch.yml'),
      patches: [],
    }
    const profileResolutionMode = options.profileResolutionMode ?? 'runtime'
    const resolutionOptions = { installAnchor: INSTALL_ANCHOR, home: harnessHome, profile }
    const resolution = profileResolutionMode === 'runtime'
      ? await createProfileResolutionGeneration(resolutionOptions)
      : await healProfilesModuleFallback(resolutionOptions)
    await mkdir(profileDir, { recursive: true })
    const rootConfig = join(profileDir, 'cordis.yml')
    await writeFile(rootConfig, '[]\n')
    ctx.baseUrl = pathToFileURL(profileDir).href + '/'
    let profileContext: ProfileContext | undefined
    if (options.profile !== undefined) {
      // A real profile: the shipped web bundles plus each fixture package,
      // installed the way `dsh plugin add` leaves them.
      const dependencies: Record<string, string> = {}
      const bundles = ['@deepseek-ai/dsh-base', '@deepseek-ai/dsh-web-app']
      for (const entry of options.profile.packages) {
        const manifest = JSON.parse(await readFile(join(entry.dir, 'package.json'), 'utf8')) as { name: string }
        dependencies[manifest.name] = `file:${entry.dir}`
        if (entry.enabled === true) bundles.push(manifest.name)
        const link = join(profileDir, 'node_modules', manifest.name)
        await mkdir(dirname(link), { recursive: true })
        await symlink(entry.dir, link, 'junction')
      }
      initProfile(profileDir, bundles)
      const manifest = readProfileManifest('dsh', profileDir)
      manifest.dependencies = dependencies
      await writeFile(join(profileDir, 'package.json'), JSON.stringify(manifest, null, 2) + '\n')
      profileContext = {
        name: 'scaffold', dir: profileDir, patchPath: profile.patchPath, installAnchor: INSTALL_ANCHOR,
        cwd: workspaceCwd, home: harnessHome, startedBundles: bundles,
        overlays: overlayPatches, telemetryDisabledEnv: undefined,
      }
      // HMR gates file-driven reloads on application readiness, which the
      // launcher commits after boot; this direct harness is ready at once.
      ctx.provide('appReady', { onReady: (listener) => { listener(); return () => {} } })
      ctx.provide('profileContext', profileContext)
    }
    // This direct Loader harness supplies the same root-path capability as app-boot.
    ctx.provide('dshHomePath', dshHomePath)
    // A host with no command line still provides one: the web bundle's startup
    // row releases the rows waiting on it, and with no arguments each starts on
    // the values this scaffold composed above. An exit request can only come
    // from a rejected argument, which a fixed empty list has none of.
    provideCmdline(ctx, {
      args: [],
      exit: (code) => {
        throw new Error(`web e2e scaffold: the web app requested exit ${String(code)} with no arguments to reject`)
      },
    })
    await ctx.plugin(PluginPackages, {
      generation: resolution,
      behavior: profileResolutionMode === 'dual' ? 'verify' : 'enforce',
    })
    await ctx.plugin(Loader)
    if (profileContext === undefined) {
      ctx.loader.builtins.include = Include
      // `cordis:group` beside it, exactly as `boot()` registers it: a group row is
      // how a preset gives one `isolate` realm to a provider and its consumers,
      // and a preset resolving package names from its own directory cannot reach
      // `@deepseek-ai/cordis-plugin-group` by name.
      ctx.loader.builtins.group = Group
      await ctx.loader.create({
        name: 'cordis:include',
        config: { path: pathToFileURL(rootConfig).href, patches },
      })
    } else {
      // The launcher's own mount, so the manager's reloads find the root Include
      // and compose the same layers the profile files name; bare names still
      // resolve through the resolution generation above, as in the direct mount.
      await mountRootInclude(ctx, rootConfig, readProfilePatches('dsh', profileContext))
    }
    await ctx.loader.await()
    await auditStartupEntries(ctx, 'web e2e scaffold')
    if (options.welcomeNoticePending !== true) {
      await ctx.settings.mutate(WELCOME_NOTICE_SETTINGS_NAMESPACE, [{
        op: 'set', path: [WELCOME_NOTICE_ACK_FIELD], value: WELCOME_NOTICE_VERSION,
      }])
    }
    const boundPort = ctx.get('webServer')?.port
    if (boundPort === undefined) {
      throw new Error('web e2e scaffold: webServer service missing after settled boot')
    }
    port = boundPort

    // Fill the open llm seam on the settled root ctx. Ordinary keyless modes
    // disable the direct adapter; the first-run lane keeps the selected adapter but has no
    // replay fixture and never streams. The direct install, unlike the plugin
    // row, returns the ReplayHandle for the teardown consumption check.
    if (options.replayProvidersOnly) {
      if (replayFixture === undefined) {
        throw new Error('replayProvidersOnly requires replayFixture (its file supplies the header)')
      }
      const fixtureText = readFileSync(replayFixture, 'utf8')
      // The consumption check is skipped for this mode, so no script source
      // may carry callable entries: reject override/child sources outright
      // and any call-bearing fixture.
      if (options.replayOverride !== undefined || replayChildFixtures !== undefined) {
        throw new Error('replayProvidersOnly cannot combine with replayOverride or replayChildFixtures')
      }
      // A fixture without a session header row must not mount the catalog
      // silently: the consumption-skip assumes the header-only shape.
      let headerType: unknown
      try {
        headerType = (JSON.parse(fixtureText.trimStart().split('\n', 1)[0] ?? '') as { type?: unknown }).type
      } catch {
        headerType = undefined
      }
      if (headerType !== 'session') {
        throw new Error('replayProvidersOnly fixture must open with a session header row')
      }
      const recorded = parseSessionLog(fixtureText)
      const hasModelCall = recorded.some(event => (
        event.type === 'assistant/message' || event.type === 'assistant/attempt'
          || event.type === 'request/header' || event.type === 'tool/call'
      ))
      if (hasModelCall) {
        throw new Error('replayProvidersOnly fixture must record no model calls')
      }
    }
    if (mode !== 'record' && replayFixture !== undefined) {
      replayHandle = installLlmReplay(ctx, {
        file: replayFixture,
        providers: (options.replayProviders ?? replayProviders(options.replayContextWindow, messages)).map(provider => ({
          ...provider,
          ...(options.replayRetryPolicy === undefined ? {} : { retryPolicy: options.replayRetryPolicy }),
        })),
        ...(options.replayOverride === undefined ? {} : { overrideFile: options.replayOverride }),
        ...(replayChildFixtures === undefined ? {} : { childFiles: replayChildFixtures }),
        ...(options.paceMs === undefined ? {} : { paceMs: options.paceMs }),
      })
    } else if (mode !== 'record' && options.deepSeekMissingCredential !== true) {
      // No fixture and no shipped adapter would leave the tree with ZERO
      // provider routes — a state no product composition has, and one the
      // composer refuses to type into. Register the same routes
      // a fixture would, with streaming that still fails loud: the scenario
      // issues no model calls, and one that slipped in must not pass quietly.
      ctx.effect(() => ctx.llm.registerAdapter(
        replayProviders(options.replayContextWindow, messages).map(provider => provider.id),
        new RouteOnlyAdapter(replayProviders(options.replayContextWindow, messages)),
      ), 'web e2e scaffold: route-only adapter')
    }
    baseUrl = `http://${browserHost}:${String(port)}`
    authenticatedUrl = ctx.connection.authenticatedUrl(baseUrl)
    const login = await fetch(authenticatedUrl, { redirect: 'manual' })
    const setCookie = login.headers.get('set-cookie')
    if (login.status !== 303 || login.headers.get('location') !== '/' || setCookie === null) {
      throw new Error('web e2e scaffold: browser token exchange did not return its session cookie')
    }
    cookieHeader = setCookie.split(';', 1)[0] ?? ''
    if (cookieHeader.length === 0) {
      throw new Error('web e2e scaffold: browser token exchange returned an empty session cookie')
    }
  } catch (error) {
    if (process.cwd() !== originalCwd) process.chdir(originalCwd)
    const cleanupFailures = await cleanupScaffoldWorld(ctx, workspaceCwd, persistenceRoot)
    restoreCredentialEnvironment()
    restoreSkillRootEnvironment()
    if (cleanupFailures.length > 0) {
      throw new AggregateError([error, ...cleanupFailures], 'web scaffold setup failed and cleanup was incomplete')
    }
    throw error
  } finally {
    if (process.cwd() !== originalCwd) process.chdir(originalCwd)
  }

  return {
    harnessHome,
    mode,
    baseUrl,
    authenticatedUrl,
    ctx,
    workspaceCwd,
    persistenceRoot,
    hostFetch(path: string, init: RequestInit = {}): Promise<Response> {
      const headers = new Headers(init.headers)
      headers.set('cookie', cookieHeader)
      return fetch(new URL(path, baseUrl), { ...init, headers })
    },
    // Barrier stack: the in-process turn/end identifies the session, its
    // explicit flush makes the transcript durable, and the caller's browser
    // settled-poll comes last because host completion strictly precedes render.
    whenTurnSettled(timeoutMs = mode === 'record' ? 180_000 : 30_000): Promise<SessionId> {
      return new Promise<SessionId>((resolveSettled, reject) => {
        const timer = setTimeout(() => {
          off()
          reject(new Error(`no turn/end within ${timeoutMs}ms`))
        }, timeoutMs)
        const off = ctx.on('session/event', (session: Session, event: SessionEvent) => {
          if (event.type !== 'turn/end') return
          clearTimeout(timer)
          off()
          ctx.sessions.flush(session)
            .then(() => { resolveSettled(session.id) }, reject)
        })
      })
    },
    async close(): Promise<void> {
      const failures: unknown[] = []
      if (mode !== 'record'
        && replayFixture !== undefined
        && options.replayProvidersOnly !== true
        && compareReplaySession) {
        try {
          await assertReplaySession(
            [...observedSessions.values()],
            replayFixture,
            mode,
            `http://${browserHost}:${port}`,
            harnessHome,
          )
        } catch (error) {
          failures.push(error)
        }
      }
      // Fixture-consumption check first, while the run's binding state is
      // still authoritative — a scenario that drove fewer model calls than
      // recorded fails here instead of drifting green. Skipped for
      // replayProvidersOnly, whose fixture is validated call-free at boot.
      if (!options.replayProvidersOnly) {
        try {
          replayHandle?.assertConsumed()
        } catch (error) {
          failures.push(error)
        }
      }
      try {
        stopObservingSessions()
        failures.push(...await cleanupScaffoldWorld(ctx, workspaceCwd, persistenceRoot))
      } finally {
        restoreCredentialEnvironment()
        restoreSkillRootEnvironment()
      }
      if (failures.length > 0) throw new AggregateError(failures, 'web scaffold teardown failed')
    },
  }
}

/**
 * Serialize a live session to the canonical raw session-JSONL layout — the
 * in-memory record-mode harvest, so the on-disk zstd default never matters.
 */
function rawSessionLog(session: Session): string {
  const encodedEvents = (session.snapshotEvents() as unknown as readonly SessionFormatEvent[])
    .map(event => sessionFormatCatalog.encodeCurrentEvent(event))
  const header = sessionFormatCatalog.encodeCurrentHeader({
    ...session.header,
    delegationDepth: session.header.delegationDepth ?? 0,
  }, session.inheritedEventCount)
  return [
    JSON.stringify(header),
    ...encodedEvents.map(record => JSON.stringify(record)),
    '',
  ].join('\n')
}

function mapJsonStringValues(value: unknown, map: (value: string) => string): unknown {
  if (typeof value === 'string') return map(value)
  if (Array.isArray(value)) return value.map(item => mapJsonStringValues(item, map))
  if (value !== null && typeof value === 'object') {
    return Object.fromEntries(Object.entries(value).map(([key, item]) => [
      key,
      mapJsonStringValues(item, map),
    ]))
  }
  return value
}

/** Tokenize the browser timezone carried by user message sources. */
function normalizeClientTimeZones(value: unknown): unknown {
  if (Array.isArray(value)) return value.map(item => normalizeClientTimeZones(item))
  if (value !== null && typeof value === 'object') {
    const next = Object.fromEntries(Object.entries(value).map(([key, item]) => [
      key,
      normalizeClientTimeZones(item),
    ]))
    const source = (next as { source?: unknown }).source
    if (source !== null && typeof source === 'object'
      && (source as { kind?: unknown }).kind === 'user'
      && typeof (source as { clientTimeZone?: unknown }).clientTimeZone === 'string') {
      return {
        ...next,
        source: { ...source, clientTimeZone: '{{clientTimeZone}}' },
      }
    }
    return next
  }
  return value
}

const WEB_PATH_TEXT_BOUNDARY_RE = /[\s<>'"`()\[\]{},;:!?=]/
const WEB_FILE_URI_PATH_PREFIX_RE = /(?:^|[^a-z0-9+.-])file:\/\/\/?$/i

function isWebCwdMatch(value: string, start: number, length: number): boolean {
  const before = value[start - 1]
  const after = value[start + length]
  const afterPunctuation = value[start + length + 1]
  const startsAtBoundary = before === undefined
    || WEB_PATH_TEXT_BOUNDARY_RE.test(before)
    || WEB_FILE_URI_PATH_PREFIX_RE.test(value.slice(0, start))
  const endsAtBoundary = after === undefined
    || after === '/'
    || after === '\\'
    || WEB_PATH_TEXT_BOUNDARY_RE.test(after)
    || after === '.' && (afterPunctuation === undefined || WEB_PATH_TEXT_BOUNDARY_RE.test(afterPunctuation))
  return startsAtBoundary && endsAtBoundary
}

function replaceWebCwd(value: string, cwd: string): string {
  let cursor = 0
  let normalized = ''
  while (cursor < value.length) {
    const match = value.indexOf(cwd, cursor)
    if (match < 0) return normalized + value.slice(cursor)
    const end = match + cwd.length
    if (isWebCwdMatch(value, match, cwd.length)) {
      normalized += value.slice(cursor, match) + '{{cwd}}'
      cursor = end
    } else {
      normalized += value.slice(cursor, end)
      cursor = end
    }
  }
  return normalized
}

/**
 * Normalize Web-only volatile strings while preserving JSON structure and row framing.
 * @param log - raw Session JSONL.
 * @param workspaceCwd - optional scaffold parent used before a live Session selects its cwd.
 * @returns compact JSONL with run-local strings tokenized.
 */
export function normalizeWebSessionVolatiles(log: string, workspaceCwd?: string): string {
  const headerLine = log.split(/\r?\n/).find(line => line.trim().length > 0)
  const header = headerLine === undefined ? undefined : JSON.parse(headerLine) as { cwd?: unknown }
  const sessionCwd = typeof header?.cwd === 'string' && header.cwd.length > 0 ? header.cwd : undefined
  const cwdSpellings = [...new Set([sessionCwd ?? workspaceCwd]
    .filter((value): value is string => typeof value === 'string' && value.length > 0)
    .flatMap((value) => {
      const forward = value.replaceAll('\\', '/')
      const native = /^[A-Za-z]:[\\/]/.test(value) ? forward.replaceAll('/', '\\') : value
      return [value, value.replaceAll('\\', '\\\\'), forward, native]
    }))].sort((left, right) => right.length - left.length)
  return log.split(/\r?\n/).map((line) => {
    if (line.trim() === '') return line
    const record = normalizeClientTimeZones(mapJsonStringValues(JSON.parse(line), (value) => {
      let normalized = value
        .replace(/Anonymous user: [0-9a-f-]{36}(?=\.$)/gi, 'Anonymous user: {{anonymousUserId}}')
      for (const cwd of cwdSpellings) normalized = replaceWebCwd(normalized, cwd)
      return normalized
    })) as { type?: unknown; data?: { endpoint?: unknown } }
    if (record.type === 'web/deepseek-search-llm-request' && typeof record.data?.endpoint === 'string') {
      record.data.endpoint = '{{webSearchEndpoint}}'
    }
    return JSON.stringify(record)
  }).join('\n')
}

function stableSessionFixture(
  session: Session,
  existing: string,
  workspaceCwd: string,
  harnessHome: string,
): string {
  const prepared = prepareSessionSnapshotFixtureForComparison(
    normalizeWebSessionVolatiles(rawSessionLog(session), workspaceCwd),
  )
  const stabilized = existing === ''
    ? prepared
    : stabilizeRefreshLog(prepared, existing, [], {
      sessionIds: [String(session.id)],
      cwd: workspaceCwd,
    })
  const fresh = scrubSessionSnapshot(stabilized)
    .split(session.id).join('{{session:1}}')
    .split(harnessHome).join('{{harnessHome}}')
  const stable = redactSessionSnapshotIds(stabilizeFixtureMessageIds([fresh], [existing]))[0]
  if (stable === undefined) throw new Error('session harvest produced no stabilized fixture')
  return stable
}

async function assertReplaySession(
  sessions: readonly Session[],
  fixturePath: string,
  mode: WebSnapshotMode,
  webUrl: string,
  harnessHome: string,
): Promise<void> {
  let expected = await readFile(fixturePath, 'utf8')
  const fixtureDir = dirname(fixturePath)
  const manifestPath = join(fixtureDir, 'snapshot.yml')
  const manifest = parseSnapshotManifest(await readFile(manifestPath, 'utf8'), manifestPath)
  let expectedPath = fixturePath
  const userPrompts = fixtureUserPrompts(expected)
  const candidates = sessions.filter((session) => {
    if (session.header.parentSession !== undefined) return false
    const actual = session.snapshotEvents().flatMap((event) => {
      if (event.type !== 'user/message' || event.data.source.kind !== 'user') return []
      const text = event.data.content.filter(block => block.type === 'text').map(block => block.text).join('')
      return text.length === 0 ? [] : [text]
    })
    return JSON.stringify(actual) === JSON.stringify(userPrompts)
  })
  expect(candidates, `Web replay fixture ${fixturePath} must match one live root session`).toHaveLength(1)
  const session = candidates[0] as Session
  const sessionCwd = session.header.cwd
  if (sessionCwd === undefined) throw new Error(`${fixturePath}: replayed session has no cwd`)
  const actual = rawSessionLog(session)
  if (mode === 'refresh' && writesCurrentSessionFixtures(manifest, mode)) {
    expected = stableSessionFixture(session, expected, sessionCwd, harnessHome)
    expectedPath = recordedSessionFixturePath(fixturePath, session.header.version)
    await writeFile(expectedPath, expected)
  }
  const expectedHeader = JSON.parse(expected.split('\n').find(line => line.trim() !== '') ?? '{}') as {
    id?: unknown
    cwd?: unknown
  }
  const actualContext: NormalizeContext = { sessionIds: [String(session.id)], cwd: sessionCwd }
  const expectedContext: NormalizeContext = {
    sessionIds: typeof expectedHeader.id === 'string' ? [expectedHeader.id] : [],
    cwd: typeof expectedHeader.cwd === 'string' ? expectedHeader.cwd : '\0no-cwd\0',
  }
  const actualSnapshot = normalizeSessionSnapshots([normalizeWebSessionVolatiles(actual)], actualContext)[0]
    ?.split(harnessHome).join('{{harnessHome}}')
  const expectedSnapshot = normalizeSessionSnapshots([normalizeWebSessionVolatiles(expected)], expectedContext)[0]
    ?.split(harnessHome).join('{{harnessHome}}')
  expect(actualSnapshot, `${fixturePath}: persisted replay`).toBe(expectedSnapshot)

  if (manifest.header?.pin !== true) return
  const normalizePrompt = (value: string): string => value
    .split(REPO_ROOT).join('{{sourceRoot}}')
    .split(webUrl).join('{{webUrl}}')
  const prompts = normalizedSystemPrompts(actual, actualContext).map(normalizePrompt)
  const schemas = normalizedToolSchemas(actual, actualContext)
  const promptPath = join(fixtureDir, 'system-prompt.expected.md')
  const schemaPath = join(fixtureDir, 'tool-schemas.expected.json')
  const promptSnapshot = formatSystemPromptSnapshot(prompts[0] as string, prompts.slice(1))
  const schemaSnapshot = formatToolSchemasSnapshot(schemas[0] as unknown[], schemas.slice(1))
  if (mode === 'refresh') {
    await Promise.all([writeFile(promptPath, promptSnapshot), writeFile(schemaPath, schemaSnapshot)])
  }
  expect(promptSnapshot, `${fixturePath}: system-prompt pin`).toBe(await readFile(promptPath, 'utf8'))
  expect(schemaSnapshot, `${fixturePath}: tool-schema pin`).toBe(await readFile(schemaPath, 'utf8'))
}

/**
 * Record-mode fixture write-back: harvest the live session, scrub the
 * system-prompt text to {{system}} and header tool schemas to {{tools}},
 * tokenize the run-local cwd, redact opaque identities with typed
 * relationship-preserving tokens, and write the fixture.
 * A manifest-retained historical generation makes the write-back a no-op.
 * @param scaffold - the record-mode scaffold.
 * @param sessionId - the driven session.
 * @param fixturePath - the committed session.jsonl target.
 */
export async function recordFixture(scaffold: WebScaffold, sessionId: SessionId, fixturePath: string): Promise<void> {
  const agent = scaffold.ctx.agents.get(sessionId)
  if (agent === undefined) throw new Error(`record harvest: no live agent for ${sessionId}`)
  const manifestPath = join(dirname(fixturePath), 'snapshot.yml')
  const manifest = parseSnapshotManifest(await readFile(manifestPath, 'utf8'), manifestPath)
  if (!writesCurrentSessionFixtures(manifest, 'record')) return
  const target = recordedSessionFixturePath(fixturePath, agent.session.header.version)
  const existingPath = existsSync(target) ? target : fixturePath
  const existing = existsSync(existingPath) ? await readFile(existingPath, 'utf8') : ''
  await writeFile(target, stableSessionFixture(
    agent.session,
    existing,
    scaffold.workspaceCwd,
    scaffold.harnessHome,
  ))
}

/**
 * The user prompts recorded in a fixture, in order — the single source tying
 * spec drive steps to recorded reality so script and fixture cannot drift.
 * @param fixtureText - raw session.jsonl contents.
 * @returns the recorded user prompt texts.
 */
export function fixtureUserPrompts(fixtureText: string): string[] {
  return parseSessionLog(fixtureText).flatMap((event) => {
    if (event.type !== 'user/message' || event.data.source.kind !== 'user') return []
    const text = event.data.content.filter(block => block.type === 'text').map(block => block.text).join('')
    return text.length > 0 ? [text] : []
  })
}

/** Deterministic UUID used when a seed fixture's typed identity token is materialized. */
export function fixtureIdentity(
  kind: 'message' | 'approval' | 'workflow' | 'command' | 'rpc' | 'retry' | 'id',
  ordinal: number,
): string {
  const hex = createHash('sha256').update(`${kind}:${ordinal}`).digest('hex').slice(0, 32).split('')
  hex[12] = '4'
  hex[16] = ['8', '9', 'a', 'b'][Number.parseInt(hex[16] as string, 16) % 4] as string
  return `${hex.slice(0, 8).join('')}-${hex.slice(8, 12).join('')}-${hex.slice(12, 16).join('')}-${hex.slice(16, 20).join('')}-${hex.slice(20).join('')}`
}

/**
 * Realize a recorded seed fixture against one scaffold: substitute the
 * `{{sessionId}}`/`{{cwd}}`/`{{harnessHome}}` placeholders and rewrite the
 * recorded cwd to the scaffold's workspace. Idempotent, so a caller may realize early (e.g. to
 * price content exactly as the host will fold it) and still pass the result
 * through {@link seedSession}.
 * @param scaffold - the booted scaffold whose workspace the seed targets.
 * @param fixtureText - the committed seed fixture text.
 * @param id - the session id the seed is realized for.
 * @returns the realized fixture text.
 */
export function realizeSeedFixture(scaffold: WebScaffold, fixtureText: string, id: string): string {
  const firstLine = fixtureText.split(/\r?\n/).find(line => line.trim().length > 0)
  const fixtureCwd = firstLine === undefined
    ? undefined
    : (JSON.parse(firstLine) as { cwd?: unknown }).cwd
  return fixtureText.split(/\r?\n/).map((line) => {
    if (line.trim() === '') return line
    const realized = mapJsonStringValues(JSON.parse(line), (value) => {
      let result = typeof fixtureCwd === 'string'
        ? value.split(fixtureCwd).join(scaffold.workspaceCwd)
        : value
      result = result
        .split('{{sessionId}}').join(id)
        .split('{{session:1}}').join(id)
        .replace(/\{\{session:([2-9]\d*)\}\}/g, (_token, ordinal: string) => `${id}-child-${ordinal}`)
        .replace(/\{\{(message|approval|workflow|command|rpc|retry|id):([1-9]\d*)\}\}/g, (_token, kind: string, ordinal: string) =>
          fixtureIdentity(kind as 'message' | 'approval' | 'workflow' | 'command' | 'rpc' | 'retry' | 'id', Number(ordinal)))
        .split('{{harnessHome}}').join(scaffold.harnessHome)
        .split('{{cwd}}').join(scaffold.workspaceCwd)
      return result
    })
    return JSON.stringify(realized)
  }).join('\n')
}

/**
 * Parse a committed web seed fixture through the replay reader.
 * @param fixtureText - session JSONL fixture contents.
 * @returns the current header line, parsed header, and logical events.
 */
/** Give a migrated fixture stream positive relative timing before its final wall-clock rebase. */
function spreadMigratedSeedStream(
  stream: SessionEvent<'assistant/message'>['data']['stream'],
): SessionEvent<'assistant/message'>['data']['stream'] {
  let nextTime = 0
  return stream.map((record) => {
    if ('time' in record) {
      const timed = { ...record, time: nextTime }
      nextTime += 1
      return timed
    }
    const timed = { ...record, time0: nextTime }
    nextTime += record.dt.reduce((total, delta) => total + delta, 0) + 1
    return timed
  })
}

export function parseSeedFixture(fixtureText: string): {
  headerLine: string
  header: Record<string, unknown>
  events: SessionEvent[]
} {
  const sourceHeaderLine = fixtureText.split(/\r?\n/).find(line => line.trim().length > 0)
  if (sourceHeaderLine === undefined) throw new Error('seed fixture has no session header')
  const sourceHeader = JSON.parse(sourceHeaderLine) as { version?: unknown }
  const current = prepareSessionSnapshotFixtureForComparison(fixtureText)
  const headerLine = current.split(/\r?\n/).find(line => line.trim().length > 0)
  if (headerLine === undefined) throw new Error('seed fixture has no session header')
  const header = JSON.parse(headerLine) as Record<string, unknown>
  if (header.type !== 'session') throw new Error('seed fixture must start with a session header')
  const events = parseSessionLog(current).map((event) => {
    if (sourceHeader.version === SESSION_FORMAT_VERSION) return event
    if (event.type === 'assistant/message') {
      return { ...event, data: { ...event.data, stream: spreadMigratedSeedStream(event.data.stream) } }
    }
    if (event.type === 'assistant/attempt') {
      return { ...event, data: { ...event.data, stream: spreadMigratedSeedStream(event.data.stream) } }
    }
    return event
  })
  return { headerLine, header, events }
}

/**
 * Render logical events as an envelope-free web seed fixture.
 * @param headerLine - original session header line.
 * @param events - logical session events in order.
 * @returns projected session JSONL.
 */
export function renderSeedFixture(
  headerLine: string,
  events: readonly ({ readonly seq: number; readonly time: number } & object)[],
): string {
  return [
    headerLine,
    ...events.map(({ seq: _seq, time: _time, ...event }) => JSON.stringify(event)),
    '',
  ].join('\n')
}

/** Re-anchor one projected embedded stream while preserving every intra-stream gap. */
function rebaseSeedStream(
  stream: SessionEvent<'assistant/message'>['data']['stream'],
  startAt: number,
): SessionEvent<'assistant/message'>['data']['stream'] {
  const first = stream[0]
  if (first === undefined) return stream
  const sourceStart = 'time' in first ? first.time : first.time0
  const delta = startAt - sourceStart
  return stream.map(record => 'time' in record
    ? { ...record, time: record.time + delta }
    : { ...record, time0: record.time0 + delta })
}

/** Last logical timestamp carried by an embedded Assistant stream. */
function seedStreamEnd(
  stream: SessionEvent<'assistant/message'>['data']['stream'],
): number | undefined {
  let end: number | undefined
  for (const record of stream) {
    const recordEnd = 'time' in record
      ? record.time
      : record.time0 + record.dt.reduce((total, delta) => total + delta, 0)
    end = end === undefined ? recordEnd : Math.max(end, recordEnd)
  }
  return end
}

/**
 * Seed a recorded session fixture into the scaffold's persistence root
 * through the real Session and JSONL APIs.
 * @param scaffold - the target scaffold.
 * @param fixtureText - raw recorded session.jsonl contents.
 * @param id - the seeded session id.
 * @param agentPreset - preset recorded by scenarios that assert resumed composition.
 * @param options - deterministic metadata overrides for ordering-sensitive scenarios.
 * @returns the seeded id.
 */
export async function seedSession(
  scaffold: WebScaffold,
  fixtureText: string,
  id: string,
  agentPreset?: string,
  options: { readonly createdAt?: number } = {},
): Promise<SessionId> {
  const decoded = parseSeedFixture(realizeSeedFixture(scaffold, fixtureText, id))
  const events = decoded.events
  if (events.length === 0) throw new Error('seed fixture has no events')
  const last = events[events.length - 1]!
  // An open final turn would be mutated by resume's crash repair on first
  // open; a committed seed must be a closed recording.
  if (last.type !== 'turn/end') throw new Error(`seed fixture must end in turn/end, got ${last.type}`)
  const createdAt = options.createdAt ?? Date.now() - 60_000
  const meta: SessionHeader = {
    version: SESSION_FORMAT_VERSION,
    id: SessionId(id),
    createdAt,
    isSeeded: false,
    cwd: scaffold.workspaceCwd,
    delegationDepth: 0,
    ...agentPreset === undefined ? {} : { agentPreset },
  }
  const fixtureCreatedAt = decoded.header.createdAt
  if (typeof fixtureCreatedAt !== 'number') {
    throw new Error('seed fixture requires a numeric createdAt header')
  }
  const timeAnchor = fixtureCreatedAt === 0 ? createdAt : fixtureCreatedAt
  let nextTime = timeAnchor
  const materializedEvents: SessionEvent[] = events.map((event) => {
    const time = nextTime
    if (event.type === 'assistant/message') {
      const stream = rebaseSeedStream(event.data.stream, time)
      const completedAt = Math.max(time, seedStreamEnd(stream) ?? time)
      nextTime = completedAt + 1
      return {
        ...event,
        time: completedAt,
        data: { ...event.data, stream },
      }
    }
    if (event.type === 'assistant/attempt') {
      const stream = rebaseSeedStream(event.data.stream, time)
      const completedAt = Math.max(time, seedStreamEnd(stream) ?? time)
      nextTime = completedAt + 1
      return {
        ...event,
        time: completedAt,
        data: { ...event.data, stream },
      }
    }
    nextTime = time + 1
    return { ...event, time }
  })
  await persistSeedSession(scaffold, meta, materializedEvents)
  return meta.id
}

/** Materialize one detached Session fixture through the shipped JSONL provider. */
async function persistSeedSession(
  scaffold: WebScaffold,
  meta: SessionHeader,
  events: readonly SessionEvent[],
): Promise<void> {
  const seeder = new Context()
  try {
    // Same root as the booted tree with the plugin's own default compression,
    // so the host's directory-scan list() sees one consistent encoding.
    await seeder.plugin(JsonlSessionPersistence, { root: scaffold.persistenceRoot })
    const handle = await seeder.sessionPersistence.create(meta)
    await handle.append(events)
    await handle.close()
  } finally {
    await seeder.fiber.dispose()
  }
}

/**
 * Read one stored session's physical event log through a throwaway read
 * handle. The physical log carries no synthetic closers: a resumed session
 * shows the closers the loop appended durably, and a never-resumed
 * interrupted log stays interrupted.
 * @param scaffold - the booted scaffold whose persistence holds the session.
 * @param id - the stored session to read.
 * @returns the stored events.
 */
export async function readPersistedEvents(scaffold: WebScaffold, id: SessionId): Promise<readonly SessionEvent[]> {
  const handle = await scaffold.ctx.sessionPersistence.open(id, 'read')
  try {
    return (await handle.read()).events
  } finally {
    await handle.close()
  }
}

/**
 * Normalize an aria snapshot: uuid, cwd, workspace-basename, duration,
 * decode-throughput, and path-sensitive compaction estimates collapse to
 * stable tokens.
 *
 * Throughput needs a token for the same reason durations do, and no fixture
 * can supply one: the figure divides a replayed step's output tokens by the
 * wall time the local run took to stream them, so it moves between two runs
 * on one machine (measured 69 → 70 tok/s) and swings wildly on a fast replay
 * (26333 tok/s for a 3 ms stream).
 */
/**
 * Relative-time buckets rendered by a dated row, in both dictionaries.
 *
 * Opt-in per capture: a session-tree golden asserts its own literal age (a
 * fresh row reads `now`, an older one does not), so collapsing the vocabulary
 * everywhere would delete that assertion. A region whose rows are dated from
 * live wall-clock state asks for it instead. Anchored on an aria label's
 * closing quote, where the bucket is always last.
 */
const ARIA_AGE =
  /(?:now|\d+min|\d+h|\d+d|\d+mo|\d+y|刚刚|\d+分钟|\d+小时|\d+天|\d+个月|\d+年)(?=")/g

function normalizeAria(snapshot: string, workspaceCwd: string, age: boolean): string {
  // The session heading renders the workspace's basename, not the full
  // path, so both spellings must collapse to the token.
  const base = workspaceCwd.split('/').pop()!
  return (age ? snapshot.replace(ARIA_AGE, '{{age}}') : snapshot)
    .split(workspaceCwd).join('{{cwd}}')
    .split(base).join('{{workspace}}')
    .replace(/[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}/gi, '{{uuid}}')
    // The optional space in `\d+m ?\d+s` covers both minute spellings: the
    // stats line's compact `2m42s` and the message-chrome template's `2m 42s`.
    .replace(
      /~\d+(?:y(?: \d+mo)?|mo(?: \d+d)?)|\b(?:\d+d(?: \d+h(?: \d+m \d+s)?)?|\d+h \d+m \d+s|\d+m ?\d+s|\d+(?:\.\d+)?s|\d+(?:\.\d+)?ms)\b/g,
      duration => duration.startsWith('~') ? duration : '{{duration}}',
    )
    .replace(/\b\d[\d,]*(?:\.\d+)? ms\b/g, '{{duration}}')
    .replace(
      /约\d+(?:年(?:\d+个月)?|个月(?:\d+天)?)|\d+(?:天(?:\d+小时(?:\d+分\d+秒)?)?|小时\d+分\d+秒|分\d+秒|(?:\.\d+)?秒)/g,
      duration => duration.startsWith('约') ? duration : '{{duration}}',
    )
    .replace(/\d+(?:\.\d+)?(?= tok\/s(?!\w))/g, '{{throughput}}')
    // Seeded compaction prices realized file paths, whose length differs
    // between local worktrees and CI scratch directories.
    .replace(/(Compacted \d+ history items \(~)\d+( tokens\))/g, '$1{{tokens}}$2')
    // Session summaries and Message IconActions clocks cross calendar
    // boundaries; collapse every shape so goldens stay stable across them.
    .replace(/\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z/g, '{{timestamp}}')
    .replace(/\d{4}年\d{1,2}月\d{1,2}日 \d{2}:\d{2}/g, '{{clock}}')
    .replace(/\d{1,2}月\d{1,2}日 \d{2}:\d{2}/g, '{{clock}}')
    .replace(/(?<!\d)\d{1,2}:\d{2}:\d{2}(?:\.\d+)?(?:\s*[AP]M)?(?!\d)/gi, '{{clock}}')
    .replace(/(?<!\d)\d{2}:\d{2}(?!\d)/g, '{{clock}}')
}

/**
 * Capture the region's aria snapshot at a settled milestone: poll until two
 * consecutive normalized captures are equal — a single-shot capture races the
 * last React commits.
 * @param page - the page under test.
 * @param selector - the region locator selector.
 * @param workspaceCwd - normalization input.
 * @param options - `normalizeAge` collapses relative-time buckets to `{{age}}`
 *   for a region whose rows are dated from live wall-clock state;
 *   `replacements` tokenizes scenario-owned values before generic normalization.
 * @returns the stable normalized snapshot.
 */
export async function captureStableAria(
  page: Page,
  selector: string,
  workspaceCwd: string,
  options: {
    normalizeAge?: boolean
    replacements?: readonly (readonly [value: string, token: string])[]
  } = {},
): Promise<string> {
  const region = page.locator(selector).first()
  const age = options.normalizeAge === true
  const normalize = (snapshot: string): string => {
    for (const [value, token] of options.replacements ?? []) {
      snapshot = snapshot.split(value).join(token)
    }
    return normalizeAria(snapshot, workspaceCwd, age)
  }
  let previous = normalize(await region.ariaSnapshot())
  await expect.poll(async () => {
    const current = normalize(await region.ariaSnapshot())
    const stable = current === previous
    previous = current
    return stable
  }, { timeout: 5_000, message: 'aria snapshot did not stabilize' }).toBe(true)
  return previous
}

/**
 * Capture a stable aria snapshot with every eligible Turn process expanded,
 * then restore the controls that were closed before the capture.
 * @param page - the page under test.
 * @param selector - the region locator selector.
 * @param workspaceCwd - normalization input.
 * @param options - optional user-visible state to establish before capture.
 * @returns the stable normalized expanded snapshot.
 */
export async function captureExpandedTurnProcessAria(
  page: Page,
  selector: string,
  workspaceCwd: string,
  options: { scrollToBottom?: boolean } = {},
): Promise<string> {
  const controls = page.locator('[data-turn-process]')
  const count = await controls.count()
  expect(count).toBeGreaterThan(0)
  const opened: number[] = []
  for (let index = 0; index < count; index++) {
    const control = controls.nth(index)
    if (!await control.isVisible() || await control.getAttribute('aria-expanded') === 'true') continue
    await control.click()
    opened.push(index)
  }
  try {
    if (options.scrollToBottom === true) {
      const backToBottom = page.getByRole('button', { name: 'Back to bottom', exact: true })
      const scroll = page.locator('[data-conversation-scroll]')
      await expect.poll(async () => {
        const distanceFromBottom = await scroll.evaluate((host) => {
          host.scrollTop = host.scrollHeight
          return host.scrollHeight - host.clientHeight - host.scrollTop
        })
        return Math.abs(distanceFromBottom) <= 1 && await backToBottom.count() === 0
      }, { timeout: 10_000 }).toBe(true)
    }
    return await captureStableAria(page, selector, workspaceCwd)
  } finally {
    for (const index of opened.reverse()) {
      const control = controls.nth(index)
      if (await control.getAttribute('aria-expanded') === 'true') await control.click()
    }
  }
}

/**
 * Compare a normalized golden, or rewrite it under refresh. Refresh is the
 * ONLY writer: a missing golden in replay mode fails with the healing command
 * instead of silently self-bootstrapping.
 * @param goldenPath - the committed ui.expected.md path.
 * @param actual - the stable normalized snapshot.
 * @param mode - the active snapshot mode.
 */
export async function compareOrRefreshGolden(goldenPath: string, actual: string, mode: WebSnapshotMode): Promise<void> {
  const payload = `${actual}\n`
  if (mode === 'refresh') {
    await writeFile(goldenPath, payload)
    return
  }
  if (!existsSync(goldenPath)) {
    throw new Error(`missing golden ${goldenPath} — run DSH_SNAPSHOT=refresh pnpm run test:web to generate it`)
  }
  expect(payload).toBe(await readFile(goldenPath, 'utf8'))
}

/**
 * Fixture-inventory guard: the scenario directory holds exactly the expected
 * files and every committed JSONL is a header-scrubbed, typed-redaction fixed point.
 * @param dir - the scenario snapshot directory.
 * @param expected - the exact expected file inventory.
 */
export async function assertFixtureInventory(dir: string, expected: string[]): Promise<void> {
  const entries = (await readdir(dir)).sort()
  const ownsManifest = entries.includes('snapshot.yml')
  const artifacts = entries.filter(name => name !== 'snapshot.yml')
  const roleInventory = (names: readonly string[]): string[] => [...new Set(names.map((name) => {
    const fixture = parseSessionFixtureName(name)
    return fixture === undefined ? name : sessionFixtureName(fixture.index, 0)
  }))].sort()
  expect(roleInventory(artifacts)).toEqual(roleInventory(expected))
  if (ownsManifest) {
    const manifestPath = join(dir, 'snapshot.yml')
    const manifest = parseSnapshotManifest(await readFile(manifestPath, 'utf8'), manifestPath)
    expect(manifest.profile).toBe('web')
    if (manifest.session === undefined) {
      expect(
        artifacts.some(name => parseSessionFixtureName(name)?.index === 0),
        `${dir}: session owner must carry a canonical parent Session fixture`,
      ).toBe(true)
    } else {
      expect(existsSync(resolve(dir, manifest.session.source)), `${dir}: session source`).toBe(true)
    }
  }
  for (const entry of artifacts.filter(name => name.endsWith('.jsonl'))) {
    const content = await readFile(join(dir, entry), 'utf8')
    expect(scrubModelRequestBulk(content), `${dir}/${entry} carries prompt text or tool-schema bulk`).toBe(content)
    expect(redactSessionSnapshotIds([content]), `${dir}/${entry} carries unredacted identities`).toEqual([content])
  }
}

/**
 * Console tripwires: reconnect/gap-repair self-healing or a pageerror must
 * fail the scenario, not mask a dead wire behind eventual consistency.
 * @param page - the page under test.
 * @returns live warning/pageerror collectors to assert empty at scenario end.
 */
export function watchConsole(page: Page): { warnings: string[]; pageErrors: string[] } {
  const warnings: string[] = []
  const pageErrors: string[] = []
  page.on('console', (message) => {
    const text = message.text()
    if (/connection lost|gap repair|discontinuous/i.test(text)) warnings.push(text)
  })
  page.on('pageerror', (error) => { pageErrors.push(String(error)) })
  return { warnings, pageErrors }
}

/**
 * Remove only connection-loss warnings emitted after an intentional reload.
 * Earlier warnings and all gap-repair/discontinuity warnings remain fatal.
 * @param tripwire - the live console-warning collector.
 * @param warningStart - warning count captured immediately before reloading.
 */
export function acknowledgeReloadConnectionLoss(
  tripwire: ReturnType<typeof watchConsole>,
  warningStart: number,
): void {
  const reloadWarnings = tripwire.warnings.splice(warningStart)
  tripwire.warnings.push(...reloadWarnings.filter(text => !/connection lost/i.test(text)))
}