feat(ru-fork): stats analytics panel

apps/server/src/persistence/Layers/ProjectionStatsFileCache.ts

@@ -0,0 +1,121 @@
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import * as Option from "effect/Option";
+import * as Schema from "effect/Schema";
+import * as Struct from "effect/Struct";
+import * as SqlClient from "effect/unstable/sql/SqlClient";
+import * as SqlSchema from "effect/unstable/sql/SqlSchema";
+
+import { NonNegativeInt, StatsSession } from "@t3tools/contracts";
+
+import { toPersistenceSqlError } from "../Errors.ts";
+import {
+  MarkAbsentInput,
+  StatsFileCacheRepository,
+  StatsFileCacheRow,
+  type StatsFileCacheRepositoryShape,
+} from "../Services/StatsFileCache.ts";
+
+// `present` is a 0/1 INTEGER column (SQLite has no boolean) → converted to boolean
+// by rowToCacheRow, exactly like McpCatalog's locked/enabled/trust. session_json
+// decodes into a StatsSession.
+const StatsFileCacheDbRow = StatsFileCacheRow.mapFields(
+  Struct.assign({
+    present: NonNegativeInt,
+    session: Schema.fromJsonString(StatsSession),
+  }),
+);
+type StatsFileCacheDbRow = typeof StatsFileCacheDbRow.Type;
+
+function rowToCacheRow(row: StatsFileCacheDbRow): StatsFileCacheRow {
+  return { ...row, present: row.present !== 0 };
+}
+
+const makeStatsFileCacheRepository = Effect.gen(function* () {
+  const sql = yield* SqlClient.SqlClient;
+
+  const selectRows = () =>
+    sql`
+      SELECT file_path AS "filePath", mtime_ms AS "mtimeMs", size_bytes AS "sizeBytes",
+             present, last_seen_at AS "lastSeenAt", session_json AS "session"
+      FROM stats_file_cache
+      ORDER BY file_path ASC
+    `;
+
+  const decodeRow = Schema.decodeUnknownEffect(StatsFileCacheDbRow);
+
+  // Decode each row INDIVIDUALLY: one undecodable session_json — a row written by an
+  // older StatsSession shape, or a corrupt blob — is dropped + logged rather than
+  // failing the whole read. The cache is rebuildable, so refresh recomputes any dropped
+  // file on its next scan. Without this, a single stale row wedges BOTH getSnapshot and
+  // refresh (refresh reads the cache before it can overwrite anything).
+  const listAll = () =>
+    Effect.gen(function* () {
+      const rawRows = yield* selectRows();
+      const decoded = yield* Effect.forEach(rawRows, (rawRow) =>
+        decodeRow(rawRow).pipe(
+          Effect.map((row) => Option.some(rowToCacheRow(row))),
+          Effect.catch(() => Effect.succeed(Option.none<StatsFileCacheRow>())),
+        ),
+      );
+      const rows = decoded.filter(Option.isSome).map((entry) => entry.value);
+      const droppedCount = decoded.length - rows.length;
+      if (droppedCount > 0) {
+        yield* Effect.logError("[stats] dropped undecodable cache rows (recomputed on next refresh)", {
+          dropped: droppedCount,
+        });
+      }
+      return rows;
+    }).pipe(Effect.mapError(toPersistenceSqlError("StatsFileCache.listAll")));
+
+  const upsertRow = SqlSchema.void({
+    Request: StatsFileCacheRow,
+    execute: (row) =>
+      sql`
+        INSERT INTO stats_file_cache (
+          file_path, mtime_ms, size_bytes, present, last_seen_at, session_json
+        )
+        VALUES (
+          ${row.filePath}, ${row.mtimeMs}, ${row.sizeBytes}, ${row.present ? 1 : 0},
+          ${row.lastSeenAt}, ${JSON.stringify(row.session)}
+        )
+        ON CONFLICT (file_path) DO UPDATE SET
+          mtime_ms = excluded.mtime_ms,
+          size_bytes = excluded.size_bytes,
+          present = excluded.present,
+          last_seen_at = excluded.last_seen_at,
+          session_json = excluded.session_json
+      `,
+  });
+
+  const markAbsentRows = SqlSchema.void({
+    Request: MarkAbsentInput,
+    execute: ({ filePaths, lastSeenAt }) =>
+      sql`
+        UPDATE stats_file_cache
+        SET present = 0, last_seen_at = ${lastSeenAt}
+        WHERE file_path IN ${sql.in(filePaths)}
+      `,
+  });
+
+  return {
+    listAll,
+    upsert: (row) =>
+      upsertRow(row).pipe(Effect.mapError(toPersistenceSqlError("StatsFileCache.upsert"))),
+    markAbsent: (input) =>
+      (input.filePaths.length === 0
+        ? Effect.void
+        : markAbsentRows(input)
+      ).pipe(Effect.mapError(toPersistenceSqlError("StatsFileCache.markAbsent"))),
+    removeByPaths: (filePaths) =>
+      (filePaths.length === 0
+        ? Effect.void
+        : sql`DELETE FROM stats_file_cache WHERE file_path IN ${sql.in(filePaths)}`.pipe(Effect.asVoid)
+      ).pipe(Effect.mapError(toPersistenceSqlError("StatsFileCache.removeByPaths"))),
+  } satisfies StatsFileCacheRepositoryShape;
+});
+
+export const StatsFileCacheRepositoryLive = Layer.effect(
+  StatsFileCacheRepository,
+  makeStatsFileCacheRepository,
+);

apps/server/src/persistence/Migrations.ts

@@ -44,6 +44,7 @@ import Migration0028 from "./Migrations/028_ProjectionThreadSessionInstanceId.ts
 import Migration0029 from "./Migrations/029_ProjectionThreadDetailOrderingIndexes.ts";
 import Migration0030 from "./Migrations/030_ProjectionThreadShellArchiveIndexes.ts";
 import Migration0031 from "./Migrations/031_Mcp.ts";
+import Migration0032 from "./Migrations/032_Stats.ts";
 
 /**
  * Migration loader with all migrations defined inline.
@@ -87,6 +88,7 @@ export const migrationEntries = [
   [29, "ProjectionThreadDetailOrderingIndexes", Migration0029],
   [30, "ProjectionThreadShellArchiveIndexes", Migration0030],
   [31, "Mcp", Migration0031],
+  [32, "Stats", Migration0032],
 ] as const;
 
 export const makeMigrationLoader = (throughId?: number) =>

apps/server/src/persistence/Migrations/032_Stats.ts

@@ -0,0 +1,22 @@
+import * as SqlClient from "effect/unstable/sql/SqlClient";
+import * as Effect from "effect/Effect";
+
+// ru-fork: the SINGLE migration for the Stats feature (unreleased — edit THIS file
+// rather than stacking 033+). One row per scanned chat file, keyed by absolute
+// file_path. mtime_ms + size_bytes are the change-detector (re-parse only on a
+// mismatch). session_json holds the computed StatsSession. present = 0 means the
+// source file is gone but its stats are retained (kept, never deleted).
+export default Effect.gen(function* () {
+  const sql = yield* SqlClient.SqlClient;
+
+  yield* sql`
+    CREATE TABLE IF NOT EXISTS stats_file_cache (
+      file_path     TEXT PRIMARY KEY,
+      mtime_ms      INTEGER NOT NULL,
+      size_bytes    INTEGER NOT NULL,
+      present       INTEGER NOT NULL DEFAULT 1,
+      last_seen_at  TEXT NOT NULL,
+      session_json  TEXT NOT NULL
+    )
+  `;
+});

apps/server/src/persistence/Services/StatsFileCache.ts

@@ -0,0 +1,45 @@
+import { StatsSession } from "@t3tools/contracts";
+import * as Context from "effect/Context";
+import type * as Effect from "effect/Effect";
+import * as Schema from "effect/Schema";
+
+import type { ProjectionRepositoryError } from "../Errors.ts";
+
+/** One persisted cache row: a computed StatsSession + on-disk identity. */
+export const StatsFileCacheRow = Schema.Struct({
+  filePath: Schema.String,
+  mtimeMs: Schema.Number,
+  sizeBytes: Schema.Number,
+  present: Schema.Boolean,
+  lastSeenAt: Schema.String,
+  session: StatsSession,
+});
+export type StatsFileCacheRow = typeof StatsFileCacheRow.Type;
+
+export const MarkAbsentInput = Schema.Struct({
+  filePaths: Schema.Array(Schema.String),
+  lastSeenAt: Schema.String,
+});
+export type MarkAbsentInput = typeof MarkAbsentInput.Type;
+
+/**
+ * Persistence for the per-file stats cache (one row per chat file, keyed by
+ * absolute file_path). Reads/writes the computed StatsSession; never the raw
+ * transcript content.
+ */
+export interface StatsFileCacheRepositoryShape {
+  readonly listAll: () => Effect.Effect<
+    ReadonlyArray<StatsFileCacheRow>,
+    ProjectionRepositoryError
+  >;
+  readonly upsert: (row: StatsFileCacheRow) => Effect.Effect<void, ProjectionRepositoryError>;
+  /** Flip present→0 (+ lastSeenAt) for files no longer on disk. Rows are kept. */
+  readonly markAbsent: (input: MarkAbsentInput) => Effect.Effect<void, ProjectionRepositoryError>;
+  /** Hard-delete rows by path (zero-usage "ghost" files that aren't real sessions). */
+  readonly removeByPaths: (filePaths: ReadonlyArray<string>) => Effect.Effect<void, ProjectionRepositoryError>;
+}
+
+export class StatsFileCacheRepository extends Context.Service<
+  StatsFileCacheRepository,
+  StatsFileCacheRepositoryShape
+>()("ru-fork/persistence/Services/StatsFileCache/StatsFileCacheRepository") {}

apps/server/src/ru-fork/common/cliRuntimeRoots.ts

@@ -0,0 +1,67 @@
+// @effect-diagnostics nodeBuiltinImport:off
+// ru-fork: single source for the CLI's RUNTIME base dir + transcript tree layout.
+// Both the advanced-chat transcript reader (qwen-transcript) and the global stats
+// scanner resolve `<base>/projects/<dir>/chats/` from here, so the brand-coupled
+// base-dir priority (and the QWEN_RUNTIME_DIR env name) can never drift between them.
+//
+// Mirrors qwen-code core/src/config/storage.ts (getRuntimeBaseDir / getProjectDir)
+// and utils/paths.ts (sanitizeCwd) EXACTLY so our resolved path === the CLI's path.
+//
+// Base dir priority (storage.ts:102-118):
+//   QWEN_RUNTIME_DIR env > settings.advanced.runtimeOutputDir > getGlobalQwenDir()
+// where getGlobalQwenDir() == os.homedir()/$CLI_DIR == ServerConfig.cliConfigDir.
+import * as path from "node:path";
+import * as os from "node:os";
+
+/** Directory segment under the runtime base that holds per-project transcript trees. */
+export const PROJECTS_DIRNAME = "projects";
+/** Directory segment under each project that holds the `<sessionId>.jsonl` chats. */
+export const CHATS_DIRNAME = "chats";
+
+/** 1:1 port of qwen utils/paths.ts:218 sanitizeCwd. `platform` is injectable for tests. */
+export const sanitizeCwd = (cwd: string, platform: NodeJS.Platform = os.platform()): string => {
+  const normalized = platform === "win32" ? cwd.toLowerCase() : cwd;
+  return normalized.replace(/[^a-zA-Z0-9]/g, "-");
+};
+
+/**
+ * 1:1 port of qwen storage.ts:40-67 resolveRuntimeBaseDir (tilde + relative).
+ * A relative `dir` resolves against `cwd` when a thread cwd is known (transcript
+ * reads), or against the home dir when it isn't (the global stats scan).
+ */
+export const expandRuntimeBaseDir = (dir: string, cwd?: string): string => {
+  let resolved = dir;
+  if (resolved === "~" || resolved.startsWith("~/") || resolved.startsWith("~\\")) {
+    const rest =
+      resolved === "~" ? [] : resolved.slice(2).split(/[/\\]+/).filter(Boolean);
+    resolved = path.join(os.homedir(), ...rest);
+  }
+  if (!path.isAbsolute(resolved)) {
+    resolved = path.resolve(cwd ?? os.homedir(), resolved);
+  }
+  return resolved;
+};
+
+export interface RuntimeBaseInput {
+  /** process env of THIS server (the child inherits it; QWEN_RUNTIME_DIR wins). */
+  readonly env: NodeJS.ProcessEnv;
+  /** ServerConfig.cliConfigDir = {home}/$CLI_DIR = qwen's default runtime base. */
+  readonly cliConfigDir: string;
+  /** Thread cwd when one is known (transcript reads); omitted for a global scan. */
+  readonly cwd?: string | undefined;
+  /** advanced.runtimeOutputDir from the CLI settings.json, if any. */
+  readonly runtimeOutputDirSetting?: string | undefined;
+}
+
+/** Mirror of storage.ts:102-118 priority: env > runtimeOutputDir setting > cliConfigDir. */
+export const resolveRuntimeBaseDir = (input: RuntimeBaseInput): string => {
+  const fromEnv = input.env["QWEN_RUNTIME_DIR"];
+  if (fromEnv && fromEnv.trim()) {
+    return expandRuntimeBaseDir(fromEnv.trim(), input.cwd);
+  }
+  const setting = input.runtimeOutputDirSetting?.trim();
+  if (setting) {
+    return expandRuntimeBaseDir(setting, input.cwd);
+  }
+  return input.cliConfigDir;
+};

apps/server/src/ru-fork/qwen-transcript/paths.ts

@@ -1,63 +1,26 @@
 // @effect-diagnostics nodeBuiltinImport:off
-// ru-fork: deterministic resolver for qwen's on-disk transcript file. Mirrors
-// qwen-code core/src/config/storage.ts (getRuntimeBaseDir / getProjectDir) and
-// utils/paths.ts (sanitizeCwd) EXACTLY so our resolved path === the CLI's path.
-//
-// Base dir priority (storage.ts:102-118):
-//   QWEN_RUNTIME_DIR env > settings.advanced.runtimeOutputDir > getGlobalQwenDir()
-// where getGlobalQwenDir() == os.homedir()/$CLI_DIR == ServerConfig.cliConfigDir.
-// We use `cliConfigDir` for the default so the base tracks our resolved CLI
-// config dir (the same root the skills scanner reads), never a literal ".qwen".
+// ru-fork: deterministic resolver for qwen's on-disk transcript FILE. The runtime
+// base-dir priority + sanitizeCwd live in common/cliRuntimeRoots.ts (shared with the
+// stats scanner so they cannot drift); this module only shapes the per-thread path
+// `<base>/projects/<sanitizeCwd(cwd)>/chats/<sessionId>.jsonl`.
 import * as path from "node:path";
-import * as os from "node:os";
 
-/** 1:1 port of qwen utils/paths.ts:218 sanitizeCwd. `platform` is injectable for tests. */
-export const sanitizeCwd = (cwd: string, platform: NodeJS.Platform = os.platform()): string => {
-  const normalized = platform === "win32" ? cwd.toLowerCase() : cwd;
-  return normalized.replace(/[^a-zA-Z0-9]/g, "-");
-};
+import {
+  CHATS_DIRNAME,
+  PROJECTS_DIRNAME,
+  type RuntimeBaseInput,
+  resolveRuntimeBaseDir,
+  sanitizeCwd,
+} from "../common/cliRuntimeRoots.ts";
 
-/** 1:1 port of qwen storage.ts:40-67 resolveRuntimeBaseDir (tilde + relative). */
-export const expandRuntimeBaseDir = (dir: string, cwd: string): string => {
-  let resolved = dir;
-  if (resolved === "~" || resolved.startsWith("~/") || resolved.startsWith("~\\")) {
-    const rest =
-      resolved === "~" ? [] : resolved.slice(2).split(/[/\\]+/).filter(Boolean);
-    resolved = path.join(os.homedir(), ...rest);
-  }
-  if (!path.isAbsolute(resolved)) {
-    resolved = path.resolve(cwd, resolved);
-  }
-  return resolved;
-};
-
-export interface TranscriptBaseInput {
-  /** process env of THIS server (the child inherits it; QWEN_RUNTIME_DIR wins). */
-  readonly env: NodeJS.ProcessEnv;
-  /** ServerConfig.cliConfigDir = {home}/$CLI_DIR = qwen's default runtime base. */
-  readonly cliConfigDir: string;
-  /** thread cwd (worktreePath ?? workspaceRoot). */
+export interface TranscriptBaseInput extends RuntimeBaseInput {
+  /** thread cwd (worktreePath ?? workspaceRoot) — always known for a transcript read. */
   readonly cwd: string;
-  /** advanced.runtimeOutputDir from the CLI settings.json, if any. */
-  readonly runtimeOutputDirSetting: string | undefined;
 }
 
-/** Mirror of storage.ts:102-118 priority. */
-export const resolveTranscriptBaseDir = (input: TranscriptBaseInput): string => {
-  const fromEnv = input.env["QWEN_RUNTIME_DIR"];
-  if (fromEnv && fromEnv.trim()) {
-    return expandRuntimeBaseDir(fromEnv.trim(), input.cwd);
-  }
-  const setting = input.runtimeOutputDirSetting?.trim();
-  if (setting) {
-    return expandRuntimeBaseDir(setting, input.cwd);
-  }
-  return input.cliConfigDir;
-};
-
 /** `<base>/projects/<sanitizeCwd(cwd)>/chats`. */
 export const resolveChatsDir = (input: TranscriptBaseInput): string =>
-  path.join(resolveTranscriptBaseDir(input), "projects", sanitizeCwd(input.cwd), "chats");
+  path.join(resolveRuntimeBaseDir(input), PROJECTS_DIRNAME, sanitizeCwd(input.cwd), CHATS_DIRNAME);
 
 /** `<chatsDir>/<sessionId>.jsonl`. */
 export const resolveTranscriptFilePath = (

apps/server/src/ru-fork/stats/StatsLayers.ts

@@ -0,0 +1,13 @@
+// ru-fork: Stats layer composition. The scanner + its file-cache repo. The repo's
+// SqlClient and the scanner's FileSystem/Path/ServerConfig come from the shared
+// server runtime graph (where this layer is provideMerged, next to MCP) — Stats is
+// DB-backed, so it registers like McpRuntimeServicesLive, not like the pure-fs
+// QwenTranscriptLive.
+import * as Layer from "effect/Layer";
+
+import { StatsFileCacheRepositoryLive } from "../../persistence/Layers/ProjectionStatsFileCache.ts";
+import { StatsScannerLive } from "./StatsScanner.ts";
+
+export const StatsLive = StatsScannerLive.pipe(
+  Layer.provide(StatsFileCacheRepositoryLive),
+);