feat(server): Retrieval Trace 基盤を追加する

server/src/__tests__/helpers/test-db.ts

@@ -160,6 +160,18 @@ export const createTestDb = async () => {
       created_at TEXT NOT NULL
     );
 
+    CREATE TABLE IF NOT EXISTS retrieval_runs (
+      id TEXT PRIMARY KEY,
+      answer_run_id TEXT,
+      thread_id TEXT,
+      message_id TEXT,
+      turn_index INTEGER,
+      query TEXT NOT NULL,
+      hits TEXT NOT NULL,
+      duration_ms INTEGER,
+      created_at TEXT NOT NULL
+    );
+
     -- 週次レポート
     CREATE TABLE IF NOT EXISTS weekly_reports (
       id TEXT PRIMARY KEY,

server/src/db/migrations/0028_retrieval_runs.sql

@@ -0,0 +1,15 @@
+CREATE TABLE retrieval_runs (
+  id TEXT PRIMARY KEY,
+  answer_run_id TEXT,
+  thread_id TEXT,
+  message_id TEXT,
+  turn_index INTEGER,
+  query TEXT NOT NULL,
+  hits TEXT NOT NULL,
+  duration_ms INTEGER,
+  created_at TEXT NOT NULL
+);
+
+CREATE INDEX idx_retrieval_runs_answer_run ON retrieval_runs(answer_run_id);
+CREATE INDEX idx_retrieval_runs_thread ON retrieval_runs(thread_id, turn_index);
+CREATE INDEX idx_retrieval_runs_created_at ON retrieval_runs(created_at);

server/src/db/schema.ts

@@ -214,6 +214,21 @@ export const llmUsage = sqliteTable("llm_usage", {
 export type LlmUsage = typeof llmUsage.$inferSelect;
 export type NewLlmUsage = typeof llmUsage.$inferInsert;
 
+export const retrievalRuns = sqliteTable("retrieval_runs", {
+  id: text("id").primaryKey(),
+  answerRunId: text("answer_run_id"),
+  threadId: text("thread_id"),
+  messageId: text("message_id"),
+  turnIndex: integer("turn_index"),
+  query: text("query").notNull(),
+  hits: text("hits").notNull(), // JSON（RetrievalHit[]）
+  durationMs: integer("duration_ms"),
+  createdAt: text("created_at").notNull(),
+});
+
+export type RetrievalRun = typeof retrievalRuns.$inferSelect;
+export type NewRetrievalRun = typeof retrievalRuns.$inferInsert;
+
 // 週次レポート（数値集計 + LLM ハイライト要約の恒久記録）
 export const weeklyReports = sqliteTable("weekly_reports", {
   id: text("id").primaryKey(),

server/src/lib/wait-until.ts

@@ -0,0 +1,7 @@
+// cloudflare:workers は workerd 専用モジュール。top-level import すると
+// このファイルを import 経由で読む Node 実行の eval スクリプトが落ちる
+export const waitUntilInBackground = (promise: Promise<unknown>) => {
+  void import("cloudflare:workers")
+    .then(({ waitUntil }) => waitUntil(promise))
+    .catch(() => {});
+};

server/src/mastra/request-context.ts

@@ -47,6 +47,8 @@ export const createRequestContext = (values: MastraRequestContextType) => {
   if (values.usageTurnIndex !== undefined) {
     requestContext.set("usageTurnIndex", values.usageTurnIndex);
   }
+  requestContext.set("answerRunId", crypto.randomUUID());
+  requestContext.set("retrievalTracePending", []);
   if (values.voiceFindings) {
     requestContext.set("voiceFindings", values.voiceFindings);
   }

server/src/repository/retrieval-run-repository.test.ts

@@ -0,0 +1,68 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+import { createTestDb, type TestDb } from "~/__tests__/helpers/test-db";
+import { retrievalRuns } from "~/db";
+
+const { testDbHolder } = vi.hoisted(() => ({
+  testDbHolder: { db: null as TestDb | null },
+}));
+
+vi.mock("~/db", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("~/db")>();
+  return {
+    ...actual,
+    createDb: () => testDbHolder.db,
+  };
+});
+
+const { retrievalRunRepository } = await import("./retrieval-run-repository");
+
+const d1 = {} as D1Database;
+
+const run = (id: string, createdAt: string) => ({
+  id,
+  answerRunId: `ar-${id}`,
+  threadId: "t-1",
+  messageId: "m-1",
+  turnIndex: 1,
+  query: "村営バスの時刻",
+  hits: "[]",
+  durationMs: 100,
+  createdAt,
+});
+
+let db: TestDb;
+
+beforeEach(async () => {
+  db = await createTestDb();
+  testDbHolder.db = db;
+});
+
+describe("deleteCreatedBefore", () => {
+  it("期限より前の検索記録だけを削除する", async () => {
+    await db
+      .insert(retrievalRuns)
+      .values([
+        run("old", "2029-01-01T00:00:00Z"),
+        run("new", "2031-01-01T00:00:00Z"),
+      ]);
+
+    const deleted = await retrievalRunRepository.deleteCreatedBefore(
+      d1,
+      "2030-01-01T00:00:00Z",
+    );
+
+    expect(deleted).toBe(1);
+    const remaining = await db.select().from(retrievalRuns).all();
+    expect(remaining.map((r) => r.id)).toEqual(["new"]);
+  });
+
+  it("対象が無ければ 0 を返す", async () => {
+    const deleted = await retrievalRunRepository.deleteCreatedBefore(
+      d1,
+      "2030-01-01T00:00:00Z",
+    );
+
+    expect(deleted).toBe(0);
+  });
+});

server/src/repository/retrieval-run-repository.ts

@@ -0,0 +1,16 @@
+import { sql } from "drizzle-orm";
+
+import { createDb, retrievalRuns } from "~/db";
+import { deleteWithCount } from "./delete-with-count";
+
+export const retrievalRunRepository = {
+  async deleteCreatedBefore(d1: D1Database, cutoff: string) {
+    const db = createDb(d1);
+
+    return deleteWithCount(
+      db,
+      retrievalRuns,
+      sql`datetime(${retrievalRuns.createdAt}) < datetime(${cutoff})`,
+    );
+  },
+};

server/src/routes/threads/chat.test.ts

@@ -53,6 +53,14 @@ vi.mock("~/mastra/request-context", () => ({
   createRequestContext: vi.fn(() => ({})),
 }));
 
+const { mockLinkRetrievalRunsToMessage } = vi.hoisted(() => ({
+  mockLinkRetrievalRunsToMessage: vi.fn(),
+}));
+
+vi.mock("~/services/knowledge/retrieval-trace", () => ({
+  linkRetrievalRunsToMessage: mockLinkRetrievalRunsToMessage,
+}));
+
 vi.mock("@mastra/memory", () => ({
   Memory: vi.fn(function () {
     return {
@@ -269,6 +277,49 @@ describe("chatRoutes: POST /:threadId/chat", () => {
     );
   });
 
+  it("ストリーム完了時に assistant メッセージ ID を retrieval run へ紐付ける", async () => {
+    useAnonAuth();
+    mockGetThreadById.mockResolvedValue(ownThread);
+
+    await routes.request(
+      buildReq("/thread-1/chat", validBody),
+      undefined,
+      mockEnv,
+    );
+
+    const { onFinish } = mockHandleChatStream.mock.calls[0][0].params;
+    onFinish({
+      totalUsage: { inputTokens: 1 },
+      response: {
+        uiMessages: [
+          { id: "user-1", role: "user" },
+          { id: "assistant-1", role: "assistant" },
+        ],
+      },
+    });
+
+    expect(mockLinkRetrievalRunsToMessage).toHaveBeenCalledWith(
+      expect.anything(),
+      "assistant-1",
+    );
+  });
+
+  it("assistant メッセージが無ければ紐付けしない", async () => {
+    useAnonAuth();
+    mockGetThreadById.mockResolvedValue(ownThread);
+
+    await routes.request(
+      buildReq("/thread-1/chat", validBody),
+      undefined,
+      mockEnv,
+    );
+
+    const { onFinish } = mockHandleChatStream.mock.calls[0][0].params;
+    onFinish({ totalUsage: { inputTokens: 1 } });
+
+    expect(mockLinkRetrievalRunsToMessage).not.toHaveBeenCalled();
+  });
+
   it("widget- prefix の resourceId では usage を platform=widget で記録する", async () => {
     useAnonAuth(WIDGET_RES_ID);
     mockGetThreadById.mockResolvedValue({

server/src/routes/threads/chat.ts

@@ -15,6 +15,7 @@ import type { ThreadVariables } from "~/middleware/require-thread-access";
 import { requireThreadAccess } from "~/middleware/require-thread-access";
 import { widgetSiteRepository } from "~/repository/widget-site-repository";
 import { nextTurnIndex, recordLlmUsage } from "~/services/analytics/llm-usage";
+import { linkRetrievalRunsToMessage } from "~/services/knowledge/retrieval-trace";
 
 export const chatRoutes = new OpenAPIHono<{
   Bindings: CloudflareBindings;
@@ -34,6 +35,16 @@ const ChatSendRequestSchema = z.object({
   isGreeting: z.boolean().optional(),
 });
 
+const extractAssistantMessageId = (event: unknown) => {
+  const uiMessages =
+    (
+      event as {
+        response?: { uiMessages?: Array<{ id?: string; role?: string }> };
+      }
+    ).response?.uiMessages ?? [];
+  return uiMessages.findLast((message) => message.role === "assistant")?.id;
+};
+
 const chatRoute = createRoute({
   method: "post",
   path: "/{threadId}/chat",
@@ -153,7 +164,7 @@ chatRoutes.openapi(chatRoute, async (c) => {
           thread: threadId,
         },
     // onFinish はレスポンス返却後に発火するため waitUntil で記録を完了させる
-    onFinish: (event) =>
+    onFinish: (event) => {
       waitUntil(
         recordLlmUsage(c.env.DB, {
           model: event.model?.modelId ?? primaryModelId(modelConfig),
@@ -166,6 +177,13 @@ chatRoutes.openapi(chatRoute, async (c) => {
           turnIndex,
           durationMs: Date.now() - startedAt,
         }),
-      ),
+      );
+      const assistantMessageId = extractAssistantMessageId(event);
+      if (assistantMessageId) {
+        waitUntil(
+          linkRetrievalRunsToMessage(requestContext, assistantMessageId),
+        );
+      }
+    },
   });
 });

server/src/services/analytics/llm-usage.ts

@@ -2,6 +2,7 @@ import type { RequestContext } from "@mastra/core/request-context";
 import type { MastraOnFinishCallbackArgs } from "@mastra/core/stream";
 import { calcCostUsd, type LlmServiceTier } from "~/lib/llm-pricing";
 import { logger } from "~/lib/logger";
+import { waitUntilInBackground } from "~/lib/wait-until";
 import { llmUsageRepository } from "~/repository/llm-usage-repository";
 
 export type LlmUsagePlatform = "web" | "line" | "lp" | "widget" | "voice";
@@ -183,11 +184,7 @@ export const usageRecordingOptions =
           durationMs: Date.now() - startedAt,
           ...contextAttributes(requestContext),
         });
-        // cloudflare:workers は workerd 専用モジュール。top-level import すると
-        // このファイルを import 経由で読む Node 実行の eval スクリプトが落ちる
-        void import("cloudflare:workers")
-          .then(({ waitUntil }) => waitUntil(recording))
-          .catch(() => {});
+        waitUntilInBackground(recording);
         return recording;
       },
     };

server/src/services/data-retention.test.ts

@@ -10,6 +10,7 @@ import {
   messageFeedback,
   pollSubmissions,
   polls,
+  retrievalRuns,
   threadPersonaStatus,
 } from "~/db";
 
@@ -252,6 +253,29 @@ describe("runDataRetention", () => {
     expect(remaining.map((r) => r.id)).toEqual(["fresh"]);
   });
 
+  it("retrieval_runs のうち 90 日より前のものだけ削除する", async () => {
+    const db = testDbHolder.db as TestDb;
+    await db.insert(retrievalRuns).values([
+      {
+        id: "old",
+        query: "q",
+        hits: "[]",
+        createdAt: daysAgo(91),
+      },
+      {
+        id: "fresh",
+        query: "q",
+        hits: "[]",
+        createdAt: daysAgo(1),
+      },
+    ]);
+
+    await runDataRetention(env, { now: NOW });
+
+    const remaining = await db.select().from(retrievalRuns);
+    expect(remaining.map((r) => r.id)).toEqual(["fresh"]);
+  });
+
   it("data_retention_logs のうち 1095 日より前のものを自身も削除する", async () => {
     const db = testDbHolder.db as TestDb;
     await db.insert(dataRetentionLogs).values([
@@ -320,6 +344,7 @@ describe("runDataRetention", () => {
         "message_feedback",
         "llm_usage",
         "poll_submissions",
+        "retrieval_runs",
         "data_retention_logs",
       ]),
     );
@@ -337,14 +362,15 @@ describe("runDataRetention", () => {
       { table: "message_feedback", deletedCount: 0 },
       { table: "llm_usage", deletedCount: 0 },
       { table: "poll_submissions", deletedCount: 0 },
+      { table: "retrieval_runs", deletedCount: 0 },
       { table: "data_retention_logs", deletedCount: 0 },
     ]);
   });
 
   it("options を省略すると現在時刻を基準に実行する", async () => {
     const results = await runDataRetention(env);
 
-    expect(results).toHaveLength(8);
+    expect(results).toHaveLength(9);
     expect(results.every((r) => r.deletedCount === 0)).toBe(true);
   });
 

server/src/services/data-retention.ts

@@ -7,6 +7,7 @@ import { mastraMessageRepository } from "~/repository/mastra-message-repository"
 import { mastraResourceRepository } from "~/repository/mastra-resource-repository";
 import { mastraThreadRepository } from "~/repository/mastra-thread-repository";
 import { pollRepository } from "~/repository/poll-repository";
+import { retrievalRunRepository } from "~/repository/retrieval-run-repository";
 import { threadPersonaStatusRepository } from "~/repository/thread-persona-status-repository";
 
 const RETENTION_DAYS = {
@@ -16,6 +17,7 @@ const RETENTION_DAYS = {
   message_feedback: 180,
   llm_usage: 180, // 週次レポートに集計が恒久保存されるため raw は短期で良い
   poll_submissions: 365,
+  retrieval_runs: 90,
   data_retention_logs: 1095,
 } as const;
 
@@ -103,6 +105,14 @@ export const runDataRetention = async (
       ),
     });
 
+    results.push({
+      table: "retrieval_runs",
+      deletedCount: await retrievalRunRepository.deleteCreatedBefore(
+        env.DB,
+        cutoff(now, RETENTION_DAYS.retrieval_runs),
+      ),
+    });
+
     results.push({
       table: "data_retention_logs",
       deletedCount: await dataRetentionLogRepository.deleteExecutedBefore(