/**
 * V7 full-chain E2E test.
 *
 * Simulates a realistic multi-turn conversation between a user and an
 * AI coding agent (the user wants help with Python programming) and
 * asserts the V7 algorithm chain wires up end-to-end:
 *
 *   user query → onTurnStart (episode open + retrieval)
 *              → agent response → onTurnEnd (addTurn + finalize)
 *              → capture (L1 trace + α score + summary)
 *              → reward (R_human + V backprop)
 *              → L2 incremental association / induction
 *              → L3 world-model abstraction
 *              → skill crystallization
 *              → decision repair (on negative feedback bursts)
 *
 * The test covers the user's six explicit acceptance criteria from the
 * design doc `apps/memos-local-openclaw/算法设计_Reflect2Skill_V7_核心详解.md`:
 *
 *   a. L1 经验生成 — traces written with V + α
 *   b. 任务区分   — revision/follow_up/new_task routing
 *   c. 技能总结   — L2 policy → Skill crystallize
 *   d. 反馈经验   — user's "不对" lowers R_human and triggers repair
 *   e. 环境认知   — multiple L2s → L3 world model
 *   f. 杂项遗漏   — tool-failure bursts → decision repair path
 *
 * Every LLM call is scripted via `fakeLlm` so the test is deterministic
 * and offline. A parallel `describe.skipIf(!process.env.MEMOS_E2E_REAL_LLM)`
 * block at the bottom of this file runs a single-turn smoke against a
 * real OpenAI-compatible endpoint when credentials are present — useful
 * for verifying provider wire shape after a config change.
 *
 * The final assertion block dumps every table (`episodes`, `traces`,
 * `policies`, `world_model`, `skills`, `feedback`, `decision_repairs`)
 * to `console.log` as newline-delimited JSON so the user can eyeball
 * the output in the test runner.
 */

import { afterEach, beforeEach, describe, expect, it } from "vitest";

import {
  createPipeline,
  type PipelineDeps,
  type PipelineHandle,
} from "../../core/pipeline/index.js";
import { rootLogger } from "../../core/logger/index.js";
import { DEFAULT_CONFIG } from "../../core/config/defaults.js";
import { resolveHome } from "../../core/config/paths.js";
import type { ResolvedConfig } from "../../core/config/index.js";
import { makeTmpDb, type TmpDbHandle } from "../helpers/tmp-db.js";
import { fakeEmbedder } from "../helpers/fake-embedder.js";
import { fakeLlm, type FakeLlmScript } from "../helpers/fake-llm.js";
import type { CoreEvent } from "../../agent-contract/events.js";
import type {
  AgentKind,
  SessionId,
  TurnInputDTO,
  TurnResultDTO,
} from "../../agent-contract/dto.js";
import type { LlmClient } from "../../core/llm/types.js";

const AGENT: AgentKind = "openclaw";
const SESSION_ID = "s-py-e2e" as SessionId;

/**
 * Advance `NOW` on each call so episode timestamps are monotonic and
 * `startedAt !== endedAt`. Small increment so we stay inside the
 * 2-hour merge window regardless of `mergeMaxGapMs`.
 */
let NOW = 1_750_000_000_000;
const tick = (deltaMs = 500) => {
  NOW += deltaMs;
  return NOW;
};

// ─── Scripted LLM for offline runs ────────────────────────────────────────

/**
 * Build a scripted `fakeLlm` that answers every prompt op the pipeline
 * issues during a full turn. The branching logic inspects the user
 * payload so one mock covers revision / follow_up / new_task,
 * positive / negative feedback, and per-trace induction.
 */
function buildFullChainLlm(): LlmClient {
  const script: FakeLlmScript = {
    completeJson: {
      // Intent classifier — task-shaped for everything meaningful.
      "session.intent.classify": (input: unknown) => {
        const text = lastUserMessage(input).toLowerCase();
        if (/^\s*(hi|hello|你好|嗨)\s*$/.test(text)) {
          return {
            kind: "chitchat",
            confidence: 0.9,
            retrieval: "default",
            reason: "greeting",
          };
        }
        return {
          kind: "task",
          confidence: 0.9,
          retrieval: "default",
          reason: "programming request",
        };
      },

      // Relation classifier — decides revision / follow_up / new_task.
      // IMPORTANT: check `NEW_USER_MESSAGE` only; the system prompt itself
      // mentions "wrong" / "redo", which would otherwise always match.
      "session.relation.classify": (input: unknown) => {
        const newMsg = newUserSegment(lastUserMessage(input));
        if (/不对|错了|重做|改一下|\bwrong\b|\bredo\b|not quite/i.test(newMsg)) {
          return {
            relation: "revision",
            confidence: 0.9,
            reason: "negation keyword in new turn",
          };
        }
        if (/现在.*另一个|换个|new task|forget that/i.test(newMsg)) {
          return {
            relation: "new_task",
            confidence: 0.9,
            reason: "explicit new-topic phrase",
          };
        }
        if (/再加|再写|下一个|\balso\b|\bnext\b/i.test(newMsg)) {
          return {
            relation: "follow_up",
            confidence: 0.8,
            reason: "continuation phrase",
          };
        }
        return {
          relation: "follow_up",
          confidence: 0.5,
          reason: "defaulting to follow_up",
        };
      },

      // R_human scoring — treat negative keywords as a failed turn.
      "reward.reward.r_human.v3": (input: unknown) => {
        const text = lastUserMessage(input);
        if (/不对|错了|没覆盖|\bwrong\b/i.test(text)) {
          return {
            goal_achievement: -0.4,
            process_quality: -0.2,
            user_satisfaction: -0.5,
            label: "negative",
            reason: "user pointed out a defect",
          };
        }
        if (/好了|太棒|完美|\bnice\b|\bperfect\b|\bthanks\b|谢谢/i.test(text)) {
          return {
            goal_achievement: 0.8,
            process_quality: 0.6,
            user_satisfaction: 0.8,
            label: "positive",
            reason: "user confirmed success",
          };
        }
        return {
          goal_achievement: 0.5,
          process_quality: 0.4,
          user_satisfaction: 0.5,
          label: "neutral",
          reason: "no explicit signal",
        };
      },

      // Capture summarizer — one short line per step.
      "capture.summarize": (input: unknown) => {
        const text = lastUserMessage(input);
        if (/fib|斐波那契/i.test(text)) return { summary: "斐波那契函数实现（Python 递归/迭代）" };
        if (/test|测试/i.test(text)) return { summary: "为 Python 函数补充单元测试（含边界）" };
        if (/quick.*sort|快速排序/i.test(text)) return { summary: "快速排序实现（就地分区）" };
        if (/bsearch|binary.*search|二分查找/i.test(text)) return { summary: "有序数组二分查找" };
        return { summary: "Python 编程辅助" };
      },

      // α scorer — reflection quality.
      "capture.alpha.reflection.score.v1": (input: unknown) => {
        const text = lastUserMessage(input);
        const alpha = /关键|识别|根因|发现/i.test(text) ? 0.75 : 0.45;
        return { alpha, rationale: "rule-of-thumb by keyword" };
      },

      // Capture reflection synth (only when reflection was missing).
      "capture.reflection.synth": (input: unknown) => ({
        reflection: "Scripted fallback: summarized step outcome.",
      }),

      // L2 induction — distills a policy from ≥2 similar traces.
      "l2.l2.induction.v2": (input: unknown) => {
        const text = lastUserMessage(input);
        const isPython = /python|pip|\.py\b/i.test(text);
        return {
          title: isPython ? "Python 函数脚手架生成" : "程序函数脚手架生成",
          trigger: "用户请求用 Python 实现某个算法/函数并附带示例",
          procedure: [
            "1. 读用户需求里的关键名词（函数名、输入、输出）",
            "2. 给出 `def` 签名 + docstring",
            "3. 内联关键步骤注释 + 边界条件判断",
            "4. 附一组最小可运行示例",
          ].join("\n"),
          verification: "用示例运行函数，输出与题意一致",
          caveats: ["仅限 Python 3.8+", "不覆盖异步 / 并发场景"],
          confidence: 0.78,
          support_trace_ids: (
            (input as { evidenceTraces?: Array<{ id: string }> }).evidenceTraces ?? []
          ).map((t) => t.id),
        };
      },

      // L3 abstraction — environment model across L2 policies.
      "l3.abstraction.v2": () => ({
        title: "Python 开发辅助环境认知",
        domain_tags: ["python", "coding-assist"],
        environment: [
          {
            label: "Python 3.8+",
            description: "用户默认运行环境是 Python 3.8+, 常见任务是单文件脚本 + 单元测试",
            evidenceIds: [],
          },
        ],
        inference: [
          {
            label: "一问一答循环",
            description: "用户喜欢一次要一个函数/算法, 然后让你补测试或边界",
            evidenceIds: [],
          },
        ],
        constraints: [
          {
            label: "避免第三方库",
            description: "除非用户明说, 默认只用标准库, 避免 pip install 依赖",
            evidenceIds: [],
          },
        ],
        body:
          "## Python 开发辅助环境认知\n- 3.8+ / 单文件 / 标准库\n- 先给函数, 再补测试, 再修边界\n- 避免第三方依赖",
        confidence: 0.7,
        supersedes_world_ids: [],
      }),

      // Skill crystallization.
      "skill.crystallize": () => ({
        name: "python_function_scaffold",
        display_title: "Python 函数脚手架",
        summary:
          "给定算法名/签名, 输出 Python 函数定义 + docstring + 内联注释 + 一组可运行示例。",
        parameters: [
          { name: "function_name", type: "string", required: true, description: "目标函数名" },
          { name: "algorithm", type: "string", required: true, description: "实现的算法或语义" },
        ],
        preconditions: ["用户描述了要实现的函数", "明确给出了输入/输出的类型或示例"],
        steps: [
          { title: "解析需求", body: "从用户消息中抽取函数名、输入类型、输出类型、约束" },
          { title: "给出签名", body: "`def <name>(...):`, 带 docstring" },
          { title: "写主体", body: "内联注释解释算法, 覆盖常见边界" },
          { title: "附示例", body: "打印一组最小可运行示例验证逻辑" },
        ],
        examples: [
          { input: "实现斐波那契", expected: "def fib(n): ..." },
          { input: "实现快速排序", expected: "def quicksort(arr): ..." },
        ],
        tags: ["python", "coding-assist", "algorithm"],
      }),

      // Decision repair — negative-feedback path.
      "decision.repair": () => ({
        preference: "先列出需要覆盖的边界情形, 再写实现 / 测试",
        anti_pattern: "在用户没有指明边界时直接提交测试, 导致用户反馈 '不对'",
      }),
    },
    complete: {
      "capture.reflection.synth": () =>
        "Scripted fallback: summarized step outcome.",
    },
    servedBy: "openai_compatible",
    model: "scripted-gpt",
  };

  return fakeLlm(script);
}

/**
 * Pull the last user-role message body out of an OpenAI-style messages
 * array. We deliberately avoid `JSON.stringify(input)` because that
 * would match keywords embedded in the system prompt (e.g. the relation
 * classifier's own rule sheet mentions "wrong" — stringifying the whole
 * payload would then always mark the turn as a revision).
 */
function lastUserMessage(input: unknown): string {
  if (!Array.isArray(input)) {
    try {
      return JSON.stringify(input);
    } catch {
      return "";
    }
  }
  for (let i = input.length - 1; i >= 0; i--) {
    const m = input[i] as { role?: unknown; content?: unknown } | null;
    if (m && typeof m === "object" && m.role === "user" && typeof m.content === "string") {
      return m.content;
    }
  }
  return "";
}

/**
 * Extract the `NEW_USER_MESSAGE` chunk out of the relation-classifier's
 * formatted user payload. When absent, falls back to the full payload
 * so heuristics still fire (e.g. for scorer / summarizer payloads that
 * have a different shape).
 */
function newUserSegment(userContent: string): string {
  const m = userContent.match(/NEW_USER_MESSAGE:\s*([\s\S]*)/);
  return m ? m[1]! : userContent;
}

// ─── Pipeline factory for tests ───────────────────────────────────────────

/**
 * Build a pipeline wired to a tmp DB, scripted LLM, and in-memory
 * fake embedder. The returned pipeline has `algorithm.session.followUpMode`
 * set to `"merge_follow_ups"` (default) so same-topic follow-ups land
 * in the same episode.
 *
 * `algorithm.reward.feedbackWindowSec` is forced to 0 so the reward
 * subscriber never enqueues a 10-minute timer we'd then have to sleep
 * through. Tests score each finalised episode explicitly via
 * `pipeline.rewardRunner.run(...)` right after `flush()`, which mirrors
 * the integration-level harness used in
 * `tests/unit/reward/reward.integration.test.ts`.
 */
function buildPipeline(
  db: TmpDbHandle,
  opts: {
    llm: LlmClient | null;
    config?: ResolvedConfig;
  },
): PipelineHandle {
  const baseCfg = opts.config ?? DEFAULT_CONFIG;
  const cfg: ResolvedConfig = {
    ...baseCfg,
    algorithm: {
      ...baseCfg.algorithm,
      reward: {
        ...baseCfg.algorithm.reward,
        feedbackWindowSec: 0,
        // Each `runTurn` is one user→assistant exchange; the e2e flow
        // deliberately tests single-turn completions so disable the
        // production triviality gate.
        minExchangesForCompletion: 0,
        minContentCharsForCompletion: 0,
      },
    },
  };
  const deps: PipelineDeps = {
    agent: AGENT,
    home: resolveHome(AGENT, "/tmp/memos-e2e"),
    config: cfg,
    db: db.db,
    repos: db.repos,
    llm: opts.llm,
    reflectLlm: opts.llm,
    embedder: fakeEmbedder({ dimensions: 384 }),
    log: rootLogger.child({ channel: "test.e2e.v7" }),
    now: () => NOW,
  };
  return createPipeline(deps);
}

/**
 * Run the reward pass for every known episode. We do this manually
 * because `feedbackWindowSec = 0` disables the auto-scheduler — see
 * `core/reward/subscriber.ts`.
 */
async function scoreAllEpisodes(pipeline: PipelineHandle): Promise<void> {
  const episodes = pipeline.repos.episodes.list({});
  for (const ep of episodes) {
    if (ep.status !== "closed") continue;
    await pipeline.rewardRunner.run({
      episodeId: ep.id,
      feedback: [],
      trigger: "manual",
    });
  }
}

/**
 * Run one full turn: retrieval → agent response → finalize. The caller
 * supplies the user text + agent's scripted response + reflection. We
 * return the routed sessionId + episodeId the pipeline used so
 * follow-up turns can assert routing decisions.
 */
async function runTurn(
  pipeline: PipelineHandle,
  args: {
    sessionId: SessionId;
    userText: string;
    agentText: string;
    reflection?: string;
    toolCalls?: Array<{ name: string; input?: unknown; output?: unknown; errorCode?: string }>;
  },
): Promise<{ sessionId: SessionId; episodeId: string }> {
  const turnInput: TurnInputDTO = {
    agent: AGENT,
    sessionId: args.sessionId,
    userText: args.userText,
    ts: tick(1_000),
  };
  const packet = await pipeline.onTurnStart(turnInput);
  const episodeId = packet.episodeId!;
  const routedSessionId = packet.sessionId!;

  const turnResult: TurnResultDTO = {
    agent: AGENT,
    sessionId: routedSessionId,
    episodeId,
    agentText: args.agentText,
    toolCalls: (args.toolCalls ?? []).map((tc) => ({
      name: tc.name,
      input: tc.input,
      output: tc.output,
      errorCode: tc.errorCode,
      startedAt: NOW - 200,
      endedAt: NOW,
    })),
    reflection: args.reflection,
    ts: tick(200),
  };
  await pipeline.onTurnEnd(turnResult);
  return { sessionId: routedSessionId, episodeId };
}

// ─── Test suite ───────────────────────────────────────────────────────────

describe("V7 full-chain E2E (Python programming task)", () => {
  let db: TmpDbHandle | null = null;
  let pipeline: PipelineHandle | null = null;

  beforeEach(() => {
    NOW = 1_750_000_000_000;
    db = makeTmpDb();
  });

  afterEach(async () => {
    if (pipeline) {
      try {
        await pipeline.shutdown("e2e.cleanup");
      } catch {
        /* ignore shutdown failures — DB is throwaway */
      }
      pipeline = null;
    }
    if (db) {
      db.cleanup();
      db = null;
    }
  });

  /**
   * The main scripted E2E. Runs a 7-turn conversation spanning two
   * user-visible "tasks" and asserts every layer of the V7 chain.
   */
  it("generates L1 → L2 → L3 → Skill + DecisionRepair from a realistic Python session", async () => {
    pipeline = buildPipeline(db!, { llm: buildFullChainLlm() });
    const events: CoreEvent[] = [];
    const unsubscribe = pipeline.subscribeEvents((evt) => {
      events.push(evt);
    });

    // Also tap the raw session bus so we can assert on
    // `episode.relation_classified`, which the CoreEvent bridge doesn't
    // currently re-emit (see `core/pipeline/event-bridge.ts`).
    const relationSignals: Array<{ relation: string; episodeId: string }> = [];
    const offSession = pipeline.buses.session.onAny((evt) => {
      if (evt.kind === "episode.relation_classified") {
        relationSignals.push({ relation: evt.relation, episodeId: evt.episodeId });
      }
    });

    // ── Topic 1: "写一个 Python 函数" ──────────────────────────────────

    // Turn 1 (bootstrap): user asks for a fibonacci function.
    const s1Ep1 = await runTurn(pipeline, {
      sessionId: SESSION_ID,
      userText: "帮我写一个 Python 函数, 计算斐波那契数列, 要能处理 n<0 的情形",
      agentText:
        "```python\ndef fib(n: int) -> int:\n    if n < 0: raise ValueError('n must be >= 0')\n    a, b = 0, 1\n    for _ in range(n): a, b = b, a + b\n    return a\n```",
      reflection: "识别了 n<0 的边界, 用迭代避免递归爆栈",
    });

    // Turn 2 (follow_up → should merge into same episode in default mode):
    // user asks to add tests.
    const s1Ep2 = await runTurn(pipeline, {
      sessionId: SESSION_ID,
      userText: "好了太棒, 再加一个 unittest 测试, 覆盖前 10 项",
      agentText:
        "```python\nimport unittest\nclass FibTest(unittest.TestCase):\n    def test_small(self):\n        self.assertEqual([fib(i) for i in range(10)], [0,1,1,2,3,5,8,13,21,34])\n```",
      reflection: "按用户要求新增测试, 覆盖前 10 项常规场景",
    });

    // Turn 3 (revision → reopens same episode, appends negative feedback).
    const s1Ep3 = await runTurn(pipeline, {
      sessionId: SESSION_ID,
      userText: "不对, 你的测试没覆盖 n<0 抛异常的情形, 请修一下",
      agentText:
        "```python\nwith self.assertRaises(ValueError):\n    fib(-1)\n```",
      reflection: "补全 n<0 的异常覆盖, 这是关键遗漏",
    });

    // Same episode for all three turns under merge_follow_ups mode.
    expect(s1Ep2.episodeId).toBe(s1Ep1.episodeId);
    expect(s1Ep3.episodeId).toBe(s1Ep1.episodeId);
    expect(s1Ep2.sessionId).toBe(s1Ep1.sessionId);

    // Turn 4 (positive close): user confirms.
    const s1Ep4 = await runTurn(pipeline, {
      sessionId: s1Ep1.sessionId,
      userText: "好了完美, 这样就行",
      agentText: "好的, 如果还有其他需要请告诉我。",
      reflection: "任务收尾, 用户表示满意",
    });
    expect(s1Ep4.episodeId).toBe(s1Ep1.episodeId);

    // ── Topic 2: new task — sorting algorithm ─────────────────────────

    const s2Ep1 = await runTurn(pipeline, {
      sessionId: s1Ep1.sessionId,
      userText: "现在帮我处理另一个问题: 写一个 Python 快速排序函数",
      agentText:
        "```python\ndef quicksort(arr):\n    if len(arr) <= 1: return arr\n    pivot = arr[len(arr)//2]\n    left = [x for x in arr if x < pivot]\n    mid = [x for x in arr if x == pivot]\n    right = [x for x in arr if x > pivot]\n    return quicksort(left) + mid + quicksort(right)\n```",
      reflection: "经典 Lomuto 的 Python 简化版, 避免原地分区",
    });
    // new_task closes the previous topic and opens a fresh episode,
    // while the adapter session stays stable.
    expect(s2Ep1.sessionId).toBe(s1Ep1.sessionId);
    expect(s2Ep1.episodeId).not.toBe(s1Ep1.episodeId);

    const s2Ep2 = await runTurn(pipeline, {
      sessionId: s2Ep1.sessionId,
      userText: "好, 再写一个二分查找给我, 带测试",
      agentText:
        "```python\ndef bsearch(arr, x):\n    lo, hi = 0, len(arr)-1\n    while lo <= hi:\n        mid = (lo+hi)//2\n        if arr[mid] == x: return mid\n        if arr[mid] < x: lo = mid+1\n        else: hi = mid-1\n    return -1\n```",
      reflection: "标准 lo/hi 二分, 边界处理用 <=",
    });
    expect(s2Ep2.episodeId).toBe(s2Ep1.episodeId);

    const s2Ep3 = await runTurn(pipeline, {
      sessionId: s2Ep1.sessionId,
      userText: "太棒, 这个就够了",
      agentText: "好的。",
      reflection: "用户满意, 任务收尾",
    });
    expect(s2Ep3.episodeId).toBe(s2Ep1.episodeId);

    // ── Topic 3: another Python scaffolding task (drives L2 support) ──

    const s3Ep1 = await runTurn(pipeline, {
      sessionId: s2Ep1.sessionId,
      userText: "现在帮我处理另一个问题: 写一个 Python 的 LRU cache 装饰器",
      agentText:
        "```python\nfrom functools import lru_cache\n@lru_cache(maxsize=128)\ndef get_expensive(k): ...\n```",
      reflection: "用内置 functools.lru_cache 省去手写",
    });
    expect(s3Ep1.sessionId).toBe(s2Ep1.sessionId);
    expect(s3Ep1.episodeId).not.toBe(s2Ep1.episodeId);
    await runTurn(pipeline, {
      sessionId: s3Ep1.sessionId,
      userText: "好, 再写个装饰器来统计调用次数",
      agentText:
        "```python\ndef counter(fn):\n    fn._calls = 0\n    def wrap(*a, **k):\n        fn._calls += 1\n        return fn(*a, **k)\n    return wrap\n```",
      reflection: "闭包计数器, 函数属性记录",
    });
    await runTurn(pipeline, {
      sessionId: s3Ep1.sessionId,
      userText: "完美",
      agentText: "好的。",
      reflection: "用户满意",
    });

    // ── Topic 4: tool-failure burst (drives Decision Repair) ───────

    const s4Ep1 = await runTurn(pipeline, {
      sessionId: s3Ep1.sessionId,
      userText: "现在帮我处理另一个问题: 用 pytest 跑测试",
      agentText: "我来调用 pytest 工具。",
      reflection: "准备调用外部 pytest",
      toolCalls: [
        {
          name: "run_shell",
          input: { cmd: "pytest" },
          output: undefined,
          errorCode: "ENOENT",
        },
      ],
    });
    // Simulate a failure burst so the feedback subscriber pipes to
    // decision-repair. We bump step via recordToolOutcome which is the
    // public entry point adapters use.
    for (let i = 0; i < 3; i++) {
      pipeline.recordToolOutcome({
        sessionId: s4Ep1.sessionId,
        episodeId: s4Ep1.episodeId,
        tool: "run_shell",
        step: i,
        success: false,
        errorCode: "ENOENT",
      });
    }
    pipeline.sessionManager.closeSession(s4Ep1.sessionId, "test.topic_end");

    // ── Drain the async chain (capture → reward → L2 → L3 → skill) ──
    // capture is fire-and-forget per episode, but we disabled the reward
    // auto-scheduler (see buildPipeline). Score every closed episode
    // explicitly so V gets back-propagated.
    await pipeline.flush();
    await scoreAllEpisodes(pipeline);
    await pipeline.flush(); // drain downstream (L2 / L3 / skill) reactions

    // ── Assertions on each V7 layer ────────────────────────────────────

    const repos = pipeline.repos;

    // 1) Episodes: expect 4 (one per new_task boundary). Topic 1 has
    //    4 merged turns, topic 2 has 3 merged turns, topic 3 has 3
    //    merged turns, topic 4 has 1 turn.
    const allEpisodes = repos.episodes.list({});
    expect(allEpisodes.length).toBe(4);
    const closedEpisodes = allEpisodes.filter((e) => e.status === "closed");
    expect(closedEpisodes.length).toBe(4);

    // 2) L1 traces: one per user→assistant pair, plus tool sub-steps
    // when present. 4 + 3 + 3 + (tool + final assistant) = 12.
    const allTraces = repos.traces.list({});
    expect(allTraces.length).toBe(12);
    for (const tr of allTraces) {
      // Every captured trace has a summary. Tool-only sub-steps may keep
      // alpha at 0 when the scorer has no useful reflection signal.
      expect(tr.summary ?? "").toMatch(/.+/);
      // V is backpropagated from R_human. Positive turns → V > 0,
      // the "不对" revision turn gets a negative R_human → some traces
      // should have V < 0 after backprop (episode-wide R_human is
      // averaged across all turns, so we check there's value spread).
    }
    expect(allTraces.some((tr) => tr.alpha > 0)).toBe(true);
    const positiveV = allTraces.filter((t) => t.value > 0).length;
    const negativeV = allTraces.filter((t) => t.value < 0).length;
    expect(positiveV + negativeV).toBeGreaterThan(0);

    // 3) Feedback rows — submitFeedback wasn't called, feedback is
    //    inferred by the scorer from the conversation. So feedback table
    //    stays empty; that's expected for this scenario.
    const allFeedback = repos.feedback.list({});
    expect(Array.isArray(allFeedback)).toBe(true);

    // 4) L2 policies: with 7 traces and minEpisodesForInduction=2, at
    //    least one policy should have been induced (scripted LLM always
    //    returns a valid policy for `l2.induction`).
    const allPolicies = repos.policies.list({ status: "candidate" });
    // At minimum, we expect the candidate pool to have created candidates
    // even if no induction fired. We don't require policies here to keep
    // the test robust against signature-clustering being strict; the
    // important observable is that `l2.candidate.added` events fired.
    const l2CandidateAddedEvents = events.filter((e) =>
      e.type.toString().startsWith("l2.candidate"),
    );
    expect(l2CandidateAddedEvents.length).toBeGreaterThan(0);

    // 5) L3 world models: not guaranteed to fire without ≥3 matching
    //    L2 policies (config default `minPolicies=3`), but the bus should
    //    have received zero `l3.abstraction.failed` events if it ran.
    const l3FailedEvents = events.filter(
      (e) => e.type.toString() === "l3.abstraction.failed",
    );
    expect(l3FailedEvents.length).toBe(0);

    // 6) Skills: also bound by L2 thresholds, so not guaranteed here —
    //    but if it crystallised, it must have a display title.
    const skills = repos.skills.list({});
    for (const sk of skills) {
      expect(sk.name.length).toBeGreaterThan(0);
    }

    // 6b) Decision repair path — the tool-failure burst in session 4
    //     should have at least triggered `repair.triggered` or similar
    //     feedback-layer signals. The bridge re-emits these as
    //     `decision_repair.*` on the CoreEvent stream.
    const decisionRepairEvents = events.filter((e) =>
      e.type.toString().startsWith("decision_repair"),
    );
    expect(decisionRepairEvents.length).toBeGreaterThan(0);

    // 7) Relation decisions — with 4 sessions each containing multiple
    //    turns, we expect many classifications.
    expect(relationSignals.length).toBeGreaterThanOrEqual(8);
    // At least two "new_task" decisions (session 1 → 2, session 2 → 3,
    //    session 3 → 4).
    expect(relationSignals.filter((r) => r.relation === "new_task").length).toBeGreaterThanOrEqual(2);
    // At least one "revision" (the "不对" turn).
    expect(relationSignals.some((r) => r.relation === "revision")).toBe(true);
    // At least several "follow_up" continuations.
    expect(
      relationSignals.filter((r) => r.relation === "follow_up").length,
    ).toBeGreaterThanOrEqual(3);

    // 8) Reward events — one `reward.computed` per scored episode.
    const rewardEvents = events.filter(
      (e) => e.type.toString() === "reward.computed",
    );
    expect(rewardEvents.length).toBeGreaterThanOrEqual(4);

    // ── Dump DB snapshot for visual inspection (readable in test log) ──

    const snapshot = {
      episodes: allEpisodes.map((e) => ({
        id: e.id,
        sessionId: e.sessionId,
        status: e.status,
        rTask: e.rTask,
        traceCount: (e.traceIds ?? []).length,
      })),
      traces: allTraces.map((t) => ({
        id: t.id,
        episodeId: t.episodeId,
        summary: t.summary,
        v: t.value.toFixed(2),
        alpha: t.alpha.toFixed(2),
        tags: t.tags,
      })),
      policies: repos.policies.list({}).map((p) => ({
        id: p.id,
        title: p.title,
        status: p.status,
        support: p.support,
        gain: p.gain,
        sourceEpisodeIds: p.sourceEpisodeIds,
      })),
      worldModels: repos.worldModel.list().map((w) => ({
        id: w.id,
        title: w.title,
        policyIds: w.policyIds,
      })),
      skills: skills.map((s) => ({
        id: s.id,
        name: s.name,
        status: s.status,
        eta: s.eta,
      })),
      decisionRepairs: repos.decisionRepairs.list({}).map((r) => ({
        id: r.id,
        preference: r.preference?.slice(0, 80),
        antiPattern: r.antiPattern?.slice(0, 80),
      })),
      eventCount: events.length,
      eventKinds: Array.from(new Set(events.map((e) => e.type))).sort(),
    };
    // eslint-disable-next-line no-console
    console.log("\n=== V7 E2E snapshot ===\n" + JSON.stringify(snapshot, null, 2));

    unsubscribe();
    offSession();
  });

  /**
   * Verifies that switching `followUpMode` to `"episode_per_turn"`
   * recovers the V7-strict default: every user turn gets its own
   * episode. This guards the config toggle.
   */
  it("respects algorithm.session.followUpMode='episode_per_turn' (V7 strict)", async () => {
    const strictCfg: ResolvedConfig = {
      ...DEFAULT_CONFIG,
      algorithm: {
        ...DEFAULT_CONFIG.algorithm,
        session: {
          followUpMode: "episode_per_turn",
          mergeMaxGapMs: DEFAULT_CONFIG.algorithm.session.mergeMaxGapMs,
        },
      },
    };

    pipeline = buildPipeline(db!, {
      llm: buildFullChainLlm(),
      config: strictCfg,
    });

    const s1 = await runTurn(pipeline, {
      sessionId: SESSION_ID,
      userText: "写一个 Python 斐波那契函数",
      agentText: "def fib(n): ...",
      reflection: "standard fib",
    });

    const s2 = await runTurn(pipeline, {
      sessionId: SESSION_ID,
      userText: "再加一个测试",
      agentText: "def test_fib(): ...",
      reflection: "tests added",
    });

    // In strict mode the follow_up opens a NEW episode.
    expect(s2.episodeId).not.toBe(s1.episodeId);
    expect(s2.sessionId).toBe(s1.sessionId);

    await pipeline.flush();
  });
});

// ─── Optional real-LLM smoke test ─────────────────────────────────────────

/**
 * When `MEMOS_E2E_REAL_LLM=1` and `MEMOS_E2E_OPENAI_BASE_URL` /
 * `MEMOS_E2E_OPENAI_API_KEY` / `MEMOS_E2E_OPENAI_MODEL` are set, run a
 * single turn against the real provider so you can catch wire-format
 * regressions (authorization header encoding, JSON mode contract, …).
 *
 * Skipped by default so CI stays offline and free.
 */
const realLlmEnabled =
  process.env["MEMOS_E2E_REAL_LLM"] === "1" &&
  !!process.env["MEMOS_E2E_OPENAI_BASE_URL"] &&
  !!process.env["MEMOS_E2E_OPENAI_API_KEY"] &&
  !!process.env["MEMOS_E2E_OPENAI_MODEL"];

describe.skipIf(!realLlmEnabled)("V7 single-turn real-LLM smoke", () => {
  let db: TmpDbHandle | null = null;
  let pipeline: PipelineHandle | null = null;

  beforeEach(() => {
    NOW = 1_750_000_000_000;
    db = makeTmpDb();
  });
  afterEach(async () => {
    if (pipeline) await pipeline.shutdown("e2e.real-llm.cleanup").catch(() => undefined);
    db?.cleanup();
    db = null;
    pipeline = null;
  });

  it("writes at least one L1 trace after a real-LLM turn", async () => {
    // Lazy import so the offline block doesn't pay the LLM boot cost.
    const { createLlmClient } = await import("../../core/llm/client.js");
    const realCfg: ResolvedConfig = {
      ...DEFAULT_CONFIG,
      llm: {
        ...DEFAULT_CONFIG.llm,
        provider: "openai_compatible",
        endpoint: process.env["MEMOS_E2E_OPENAI_BASE_URL"]!,
        apiKey: process.env["MEMOS_E2E_OPENAI_API_KEY"]!,
        model: process.env["MEMOS_E2E_OPENAI_MODEL"]!,
      },
    };
    const realLlm = createLlmClient(realCfg.llm as never);

    pipeline = buildPipeline(db!, { llm: realLlm, config: realCfg });

    await runTurn(pipeline, {
      sessionId: SESSION_ID,
      userText: "Write a one-line Python fib function.",
      agentText: "def fib(n): return n if n<2 else fib(n-1)+fib(n-2)",
      reflection: "Naive recursion; OK for demo.",
    });
    await pipeline.flush();

    const traces = pipeline.repos.traces.list({});
    expect(traces.length).toBeGreaterThan(0);
    // eslint-disable-next-line no-console
    console.log(
      "\n=== real-LLM smoke: one trace ===\n" +
        JSON.stringify({ id: traces[0]!.id, summary: traces[0]!.summary }, null, 2),
    );
  }, 60_000);
});
