From fc7cb5b8edb620570c2791b671cbbcefb024c804 Mon Sep 17 00:00:00 2001 From: Yaowei Zheng Date: Wed, 5 Aug 2026 00:09:15 +0800 Subject: [PATCH] perf(server): mtime-gate the scheduler tick and memoize project-config reads (#200) Co-authored-by: Claude Fable 5 --- packages/core/src/state/project-config.ts | 42 ++-- packages/server/src/app.ts | 1 + packages/server/src/http/routes/schedules.ts | 2 +- packages/server/src/internal/mtime-gate.ts | 34 ++++ packages/server/src/runtime/schedule-store.ts | 183 +++++++++++++++--- packages/server/src/runtime/scheduler.ts | 79 ++++++-- .../src/services/project-config-service.ts | 68 ++++++- packages/server/src/services/trace-index.ts | 26 +-- .../server/test/project-config-cache.test.ts | 106 ++++++++++ packages/server/test/scheduler.test.ts | 109 ++++++++++- packages/server/test/schedules.test.ts | 31 +++ 11 files changed, 596 insertions(+), 85 deletions(-) create mode 100644 packages/server/src/internal/mtime-gate.ts create mode 100644 packages/server/test/project-config-cache.test.ts diff --git a/packages/core/src/state/project-config.ts b/packages/core/src/state/project-config.ts index 8588fb9..3934709 100644 --- a/packages/core/src/state/project-config.ts +++ b/packages/core/src/state/project-config.ts @@ -246,15 +246,37 @@ function parseDefaultChat(value: unknown): ProjectChatDefaults | undefined { } /** - * Loads the Project config; returns the default config (without writing to disk) if - * `.project_config.toml` doesn't exist. Returns plaintext (masking is applied at the interface - * layer); reports a clear error when the old format (a string reference / an entry missing - * provider) is read. + * Narrows an already-parsed `.project_config.toml` table into a typed `ProjectConfig` + * (`file` is used in error messages only). Shared by `loadProjectConfig` and callers that + * hold a cached parse of the same file (the server's ProjectConfigService), so the two + * paths can never validate differently. * * The return literal below rebuilds the config from **known keys only** — any new top-level * key must be added to `ProjectConfig` AND echoed here, or a load→save round trip (the CLI * path) silently drops it. */ +export function projectConfigFromTable( + file: string, + parsed: Record, +): ProjectConfig { + const defaultModel = parseRefField(file, "default_model", parsed.default_model); + const visionModel = parseRefField(file, "vision_model", parsed.vision_model); + const defaultChat = parseDefaultChat(parsed.default_chat); + return { + ...(parsed.name !== undefined ? { name: parsed.name as string } : {}), + ...(defaultModel !== undefined ? { default_model: defaultModel } : {}), + ...(visionModel !== undefined ? { vision_model: visionModel } : {}), + ...(defaultChat !== undefined ? { default_chat: defaultChat } : {}), + models: ((parsed.models as unknown[] | undefined) ?? []).map((m) => assertModelEntry(file, m)), + }; +} + +/** + * Loads the Project config; returns the default config (without writing to disk) if + * `.project_config.toml` doesn't exist. Returns plaintext (masking is applied at the interface + * layer); reports a clear error when the old format (a string reference / an entry missing + * provider) is read. + */ export async function loadProjectConfig(root: string, projectId: string): Promise { const file = projectConfigPath(root, projectId); let raw: string; @@ -265,17 +287,7 @@ export async function loadProjectConfig(root: string, projectId: string): Promis throw err; } // Defensive: parseToml may return null/undefined for an empty file, and destructuring it would throw a TypeError. - const parsed = (parseToml(raw) ?? {}) as Record; - const defaultModel = parseRefField(file, "default_model", parsed.default_model); - const visionModel = parseRefField(file, "vision_model", parsed.vision_model); - const defaultChat = parseDefaultChat(parsed.default_chat); - return { - ...(parsed.name !== undefined ? { name: parsed.name as string } : {}), - ...(defaultModel !== undefined ? { default_model: defaultModel } : {}), - ...(visionModel !== undefined ? { vision_model: visionModel } : {}), - ...(defaultChat !== undefined ? { default_chat: defaultChat } : {}), - models: ((parsed.models as unknown[] | undefined) ?? []).map((m) => assertModelEntry(file, m)), - }; + return projectConfigFromTable(file, (parseToml(raw) ?? {}) as Record); } /** A TOML inline table for a paired reference (reuses smol-toml's string serialization, guaranteeing correct escaping). */ diff --git a/packages/server/src/app.ts b/packages/server/src/app.ts index 7fc33a1..54de097 100644 --- a/packages/server/src/app.ts +++ b/packages/server/src/app.ts @@ -262,6 +262,7 @@ export function buildAppDeps(config: ServerConfig, overrides: BuildDepsOverrides sessions: sessionsRepo, runner: manager, sessionCreator: sessionService, + projectConfig: projectConfigService, errors, notify: (userId, event) => { channels.get(userChannelKey(userId)).publish(event, "server_event"); diff --git a/packages/server/src/http/routes/schedules.ts b/packages/server/src/http/routes/schedules.ts index 7eb3a43..cf4a32c 100644 --- a/packages/server/src/http/routes/schedules.ts +++ b/packages/server/src/http/routes/schedules.ts @@ -215,7 +215,7 @@ async function upsert( if (!parsed.ok) throw badRequest(`Invalid schedule configuration: ${parsed.error}`); // At save time, verify the (provider, modelId) pair names a configured model (same rules as reconciliation) so we never persist a broken file. // The pairing rule itself is enforced by parseScheduleFile above, which rejects half a reference. - const refError = await validateScheduleModelRef(deps.config.root, projectId, parsed.def); + const refError = await validateScheduleModelRef(deps.projectConfigService, projectId, parsed.def); if (refError !== null) throw badRequest(`Invalid schedule configuration: ${refError}`); await writeScheduleFile(deps.config.root, projectId, agentId, name, raw); // Creator attribution: the API writer is the creator (falls back to the Project owner only for hand-edited files). diff --git a/packages/server/src/internal/mtime-gate.ts b/packages/server/src/internal/mtime-gate.ts new file mode 100644 index 0000000..24612a7 --- /dev/null +++ b/packages/server/src/internal/mtime-gate.ts @@ -0,0 +1,34 @@ +/** + * Shared mtime-gate primitives for the server's derived caches (Trace index, schedule + * scans, project-config reads): disk stays the single source of truth, and a cached + * value may be served only while the stat'd mtime still matches the one recorded at + * read time. External edits are therefore caught by one stat instead of a re-read. + * + * The freshness guard is the subtle part of the contract: filesystem timestamps are + * coarse (a write and a later change can land on the same tick), so an mtime younger + * than FRESH_MS is never recorded as clean — `cacheable` returns a sentinel that can + * never match a real stat, forcing re-reads until the path has been quiet. Every gate + * in the server must go through these helpers so the contract stays in one place. + */ +import fs from "node:fs/promises"; + +/** + * mtimes younger than this are treated as UNSTABLE and never cached as clean: an + * actively-written path costs one extra re-read per pass, and a gate can never wedge + * on a same-tick change. + */ +export const FRESH_MS = 2000; + +/** A gate-cacheable mtime: the real value once stable, else a sentinel that never matches. */ +export function cacheable(mtimeMs: number): number { + return Date.now() - mtimeMs > FRESH_MS ? mtimeMs : -1; +} + +/** mtimeMs of a path; null when it does not exist. */ +export async function statMtime(p: string): Promise { + try { + return (await fs.stat(p)).mtimeMs; + } catch { + return null; + } +} diff --git a/packages/server/src/runtime/schedule-store.ts b/packages/server/src/runtime/schedule-store.ts index da07ed0..355765f 100644 --- a/packages/server/src/runtime/schedule-store.ts +++ b/packages/server/src/runtime/schedule-store.ts @@ -8,7 +8,9 @@ import fs from "node:fs/promises"; import path from "node:path"; import { stringify as stringifyToml } from "smol-toml"; -import { loadProjectConfig, resolveModelRef, scheduleDir } from "@prismshadow/penguin-core"; +import { resolveModelRef, scheduleDir } from "@prismshadow/penguin-core"; +import type { ProjectConfig } from "@prismshadow/penguin-core"; +import { cacheable, statMtime } from "../internal/mtime-gate.js"; import type { ScheduleDefinition } from "./schedule-file.js"; import { parseScheduleFile, type ScheduleParseResult } from "./schedule-file.js"; @@ -18,32 +20,162 @@ export interface ScheduleFileEntry { parsed: ScheduleParseResult; } -/** List all schedule files for this Agent (a missing directory is treated as empty). */ -export async function listScheduleFiles( - root: string, - projectId: string, - agentId: string, -): Promise { - const dir = scheduleDir(root, projectId, agentId); - let names: string[]; - try { - names = await fs.readdir(dir); - } catch { - return []; +/** One cached schedule file: its parsed entry plus the stat the parse is valid for. */ +interface CachedScheduleFile { + /** Gate-cacheable mtime (a fresh mtime is stored as a never-matching sentinel). */ + mtimeMs: number; + sizeBytes: number; + entry: ScheduleFileEntry; +} + +/** One Agent's cached schedule dir: last seen dir mtime (null = did not exist) + files in listing order. */ +interface CachedScheduleDir { + mtimeMs: number | null; + files: Map; +} + +/** + * mtime-gated schedule-dir scans (the trace-index house pattern: disk stays the + * single source of truth, mtime is the change signal, fresh mtimes are never cached + * as clean, single-flight per Agent). The scheduler tick and the schedules routes + * both list through here, so the steady state costs stats instead of a full + * readdir + readFile + TOML parse of every file every 30s: + * + * - Unchanged dir mtime → the name set is intact (create/delete/rename all move it); + * one stat per known file then catches in-place edits (a content write moves only + * the FILE's mtime, never the dir's). No readdir, no reads. + * - Changed dir mtime / first look / `force` → one readdir, then re-read only the + * files whose mtime/size moved; unchanged files reuse their cached parse. + * - Missing dir → cached as such; a later create is caught by the same stat. + * + * Manual edits are picked up by the next pass because every variant moves an mtime + * this gate stats; the routes' write paths additionally force/invalidate for + * immediate effect (see Scheduler.reconcileAgent / dropEntry). + */ +export class ScheduleFileCache { + private readonly seen = new Map(); + private readonly inflight = new Map>(); + /** Test observability: gateStats = dir stat calls; dirScans = readdir passes; fileReads = schedule-file content reads. */ + readonly counters = { gateStats: 0, dirScans: 0, fileReads: 0 }; + + constructor(private readonly root: string) {} + + private keyOf(projectId: string, agentId: string): string { + return `${projectId}\0${agentId}`; } - const entries: ScheduleFileEntry[] = []; - for (const file of names.sort()) { - if (!file.endsWith(".toml")) continue; - const name = file.slice(0, -".toml".length); + + /** + * Lists this Agent's schedule files (a missing directory is treated as empty). + * `force` ignores the dir-mtime gate (write paths' immediate-effect entry); a + * forced call issued while a scan is in flight chains a fresh pass behind it. + * The returned entries are shared with the cache — callers must not mutate them. + */ + list( + projectId: string, + agentId: string, + opts: { force?: boolean } = {}, + ): Promise { + const key = this.keyOf(projectId, agentId); + const existing = this.inflight.get(key); + if (existing) { + return opts.force === true + ? existing.then(() => this.list(projectId, agentId, opts)) + : existing; + } + const run = this.scan(projectId, agentId, opts.force === true).finally(() => { + this.inflight.delete(key); + }); + this.inflight.set(key, run); + return run; + } + + /** Write-time invalidation (route delete path): the next list re-scans regardless of mtimes. */ + invalidate(projectId: string, agentId: string): void { + this.seen.delete(this.keyOf(projectId, agentId)); + } + + /** Drops every cached Agent of the Projects NOT in `live` (deleted Projects stop being ticked, so sweep here). */ + retainProjects(live: ReadonlySet): void { + for (const key of this.seen.keys()) { + if (!live.has(key.slice(0, key.indexOf("\0")))) this.seen.delete(key); + } + } + + private async scan( + projectId: string, + agentId: string, + force: boolean, + ): Promise { + const key = this.keyOf(projectId, agentId); + const dir = scheduleDir(this.root, projectId, agentId); + const cached = this.seen.get(key); + this.counters.gateStats += 1; + const dirMtime = await statMtime(dir); + if (dirMtime === null) { + this.seen.set(key, { mtimeMs: null, files: new Map() }); + return []; + } + const next: CachedScheduleDir = { mtimeMs: cacheable(dirMtime), files: new Map() }; + let names: string[]; + if (!force && cached !== undefined && cached.mtimeMs !== null && cached.mtimeMs === dirMtime) { + // Same dir mtime (sentinels never match): the name set is unchanged; only + // in-place content edits are possible, and the per-file stats below catch those. + next.mtimeMs = cached.mtimeMs; + names = [...cached.files.keys()]; + } else { + this.counters.dirScans += 1; + let listing: string[]; + try { + listing = await fs.readdir(dir); + } catch { + // Deleted between stat and readdir: treat as missing; a later pass settles it. + this.seen.set(key, { mtimeMs: null, files: new Map() }); + return []; + } + // Sort the full file names before stripping the suffix — the pre-existing listing order. + names = listing + .sort() + .filter((f) => f.endsWith(".toml")) + .map((f) => f.slice(0, -".toml".length)); + } + const entries: ScheduleFileEntry[] = []; + for (const name of names) { + const file = await this.revalidate(dir, name, cached?.files.get(name)); + if (file === null) continue; // Deleted during reconciliation: revisit next round. + next.files.set(name, file); + entries.push(file.entry); + } + this.seen.set(key, next); + return entries; + } + + /** Reuses the cached parse while the file's stat still matches; re-reads otherwise. */ + private async revalidate( + dir: string, + name: string, + cached: CachedScheduleFile | undefined, + ): Promise { + const file = path.join(dir, `${name}.toml`); + let stat: { mtimeMs: number; size: number }; + try { + stat = await fs.stat(file); + } catch { + return null; + } + if (cached && cached.mtimeMs === stat.mtimeMs && cached.sizeBytes === stat.size) return cached; + this.counters.fileReads += 1; let raw: string; try { - raw = await fs.readFile(path.join(dir, file), "utf8"); + raw = await fs.readFile(file, "utf8"); } catch { - continue; // Deleted during reconciliation: revisit next round. + return null; } - entries.push({ name, raw, parsed: parseScheduleFile(name, raw) }); + return { + mtimeMs: cacheable(stat.mtimeMs), + sizeBytes: stat.size, + entry: { name, raw, parsed: parseScheduleFile(name, raw) }, + }; } - return entries; } export async function readScheduleFile( @@ -87,6 +219,11 @@ export function serializeSchedule(fields: { return `${stringifyToml(table)}\n`; } +/** Project-config source for model-ref validation (ProjectConfigService: same semantics as core's loadProjectConfig, served from its mtime-gated cache). */ +export interface ScheduleConfigSource { + loadConfig(projectId: string): Promise; +} + /** * Config check for a schedule's model reference (shared by save and reconciliation): * the reference is a complete `(provider, model_id)` pair and must name an entry in the @@ -95,7 +232,7 @@ export function serializeSchedule(fields: { * failure), or null when the pair is configured (or there is no model reference at all). */ export async function validateScheduleModelRef( - root: string, + configs: ScheduleConfigSource, projectId: string, def: Pick, ): Promise { @@ -106,7 +243,7 @@ export async function validateScheduleModelRef( return "provider and model_id must be given together (a model reference is always a pair); omit both to use the Project's default model"; } try { - const cfg = await loadProjectConfig(root, projectId); + const cfg = await configs.loadConfig(projectId); resolveModelRef(cfg, def.modelId, def.provider); return null; } catch (err) { diff --git a/packages/server/src/runtime/scheduler.ts b/packages/server/src/runtime/scheduler.ts index 0444020..3f6ca96 100644 --- a/packages/server/src/runtime/scheduler.ts +++ b/packages/server/src/runtime/scheduler.ts @@ -13,17 +13,24 @@ * the wait only advance, they don't stack), and send once it becomes idle. * - Bound Session deleted: record an error and mark invalid; editing the file re-activates it via reconcile. * - Deleting the file removes the task; reconcile also cleans up its SQLite run state and queue entry. + * + * The periodic scan is mtime-gated (trace-index house pattern): the agents-dir + * listing and each Agent's parsed schedule files are cached and revalidated by stat, + * so an unchanged tree ticks with zero readdir/readFile — manual edits are still + * picked up next tick because every edit variant moves an mtime the gate stats. */ import { createHash } from "node:crypto"; -import { readdir } from "node:fs/promises"; +import fs from "node:fs/promises"; import { agentsDir, buildScheduledMessage, userText } from "@prismshadow/penguin-core"; import type { ProjectsRepo } from "../db/repos/projects.js"; import type { SchedulesRepo, ScheduleStateRow } from "../db/repos/schedules.js"; import type { SessionsRepo } from "../db/repos/sessions.js"; +import { cacheable, statMtime } from "../internal/mtime-gate.js"; import type { ErrorSink } from "./error-recorder.js"; import type { ScheduleDefinition } from "./schedule-file.js"; import { latestSlotAt, slotInWindow } from "./schedule-file.js"; -import { listScheduleFiles, readScheduleFile, validateScheduleModelRef } from "./schedule-store.js"; +import { ScheduleFileCache, readScheduleFile, validateScheduleModelRef } from "./schedule-store.js"; +import type { ScheduleConfigSource } from "./schedule-store.js"; import type { ScheduleServerEvent } from "../api/types.js"; /** Reconcile and fire-check interval (min period is 5m, so 30s granularity is plenty). */ @@ -65,6 +72,8 @@ export interface SchedulerDeps { sessions: SessionsRepo; runner: ScheduleTaskRunner; sessionCreator: ScheduleSessionCreator; + /** Project-config source for model-ref validation (ProjectConfigService's mtime-cached reads). */ + projectConfig: ScheduleConfigSource; errors: ErrorSink; /** Fire and send are notified over the user-level event stream (app layer binds userChannelKey). */ notify: (userId: string, event: ScheduleServerEvent) => void; @@ -94,10 +103,15 @@ export class Scheduler { /** key = `${projectId}\0${agentId}\0${name}` */ private readonly pending = new Map(); private ticking = false; + /** mtime-gated schedule-file scans (public for test observability of its counters). */ + readonly files: ScheduleFileCache; + /** Per-Project agents-dir listing gate: dir mtime → Agent ids (creating/removing an Agent dir moves it). */ + private readonly agentDirs = new Map(); constructor(private readonly deps: SchedulerDeps) { this.now = deps.now ?? (() => Date.now()); this.intervalMs = deps.intervalMs ?? TICK_INTERVAL_MS; + this.files = new ScheduleFileCache(deps.root); } /** Start: run one reconcile immediately (startup semantics: no backfill), then enter the periodic tick. */ @@ -119,9 +133,17 @@ export class Scheduler { if (this.ticking) return; this.ticking = true; try { - for (const project of this.deps.projects.listAll()) { + const projects = this.deps.projects.listAll(); + for (const project of projects) { await this.reconcileProject(project.projectId, project.ownerUserId); } + // A deleted Project simply stops being listed, so its cache entries are swept + // here (removed Agents inside a live Project are dropped by listAgentIds). + const live = new Set(projects.map((p) => p.projectId)); + this.files.retainProjects(live); + for (const projectId of this.agentDirs.keys()) { + if (!live.has(projectId)) this.agentDirs.delete(projectId); + } await this.drainQueue(); } catch (err) { this.deps.errors.record({ source: "schedule", err, code: "schedule_tick_failed" }); @@ -130,11 +152,11 @@ export class Scheduler { } } - /** Immediate-effect entry after a route write: reconcile just one Agent and drain its queue. */ + /** Immediate-effect entry after a route write: reconcile just one Agent (bypassing the mtime gate) and drain its queue. */ async reconcileAgent(projectId: string, agentId: string): Promise { const project = this.deps.projects.findById(projectId); if (!project) return; - await this.reconcileOneAgent(projectId, agentId, project.ownerUserId); + await this.reconcileOneAgent(projectId, agentId, project.ownerUserId, { force: true }); await this.drainQueue(); } @@ -145,7 +167,7 @@ export class Scheduler { ): Promise<{ entries: ScheduleEntryView[]; invalid: Array<{ name: string; error: string }> }> { const project = this.deps.projects.findById(projectId); const owner = project?.ownerUserId ?? null; - const files = await listScheduleFiles(this.deps.root, projectId, agentId); + const files = await this.files.list(projectId, agentId); const entries: ScheduleEntryView[] = []; const invalid: Array<{ name: string; error: string }> = []; for (const file of files) { @@ -154,7 +176,11 @@ export class Scheduler { continue; } // A model ref whose (provider, model_id) pair isn't in the config is treated like a parse failure: goes to invalidFiles, not scheduled. - const refError = await validateScheduleModelRef(this.deps.root, projectId, file.parsed.def); + const refError = await validateScheduleModelRef( + this.deps.projectConfig, + projectId, + file.parsed.def, + ); if (refError !== null) { invalid.push({ name: file.name, error: refError }); continue; @@ -174,10 +200,11 @@ export class Scheduler { return { entries, invalid }; } - /** For routes: state cleanup after a task is deleted. */ + /** For routes: state cleanup after a task is deleted (the unlink moved the dir mtime, but invalidate for immediate effect anyway). */ dropEntry(projectId: string, agentId: string, name: string): void { this.pending.delete(this.keyOf(projectId, agentId, name)); this.deps.repo.delete(projectId, agentId, name); + this.files.invalidate(projectId, agentId); } // ------------------------------------------------------------------------- @@ -192,22 +219,44 @@ export class Scheduler { } } - /** Enumerate Agents under a Project: scheduling only cares about Agent dirs that exist on disk (no dir → no tasks). */ + /** + * Enumerate Agents under a Project: scheduling only cares about Agent dirs that exist on + * disk (no dir → no tasks). mtime-gated: an unchanged agents dir serves the cached ids + * with one stat and no readdir (creating/removing an Agent dir moves the dir's mtime); + * removed Agents also drop their schedule-file cache entries here. + */ private async listAgentIds(projectId: string): Promise { - try { - const items = await readdir(agentsDir(this.deps.root, projectId), { withFileTypes: true }); - return items.filter((d) => d.isDirectory()).map((d) => d.name); - } catch { + const cached = this.agentDirs.get(projectId); + const mtimeMs = await statMtime(agentsDir(this.deps.root, projectId)); + if (mtimeMs === null) { + this.agentDirs.delete(projectId); return []; } + if (cached && cached.mtimeMs === mtimeMs) return cached.ids; + let ids: string[]; + try { + const items = await fs.readdir(agentsDir(this.deps.root, projectId), { + withFileTypes: true, + }); + ids = items.filter((d) => d.isDirectory()).map((d) => d.name); + } catch { + this.agentDirs.delete(projectId); + return []; + } + for (const gone of cached?.ids.filter((id) => !ids.includes(id)) ?? []) { + this.files.invalidate(projectId, gone); + } + this.agentDirs.set(projectId, { mtimeMs: cacheable(mtimeMs), ids }); + return ids; } private async reconcileOneAgent( projectId: string, agentId: string, ownerUserId: string, + opts: { force?: boolean } = {}, ): Promise { - const files = await listScheduleFiles(this.deps.root, projectId, agentId); + const files = await this.files.list(projectId, agentId, opts); for (const file of files) { if (!file.parsed.ok) { // Skip invalid files and record an error (the recorder dedups within a short window, so storms don't spam). @@ -223,7 +272,7 @@ export class Scheduler { // At reconcile time, check the (provider, model_id) pair names a configured model: a // reference that doesn't is treated like an invalid file — skip scheduling and record an // error (recorder dedups in a short window); it recovers once the file/config is fixed. - const refError = await validateScheduleModelRef(this.deps.root, projectId, def); + const refError = await validateScheduleModelRef(this.deps.projectConfig, projectId, def); if (refError !== null) { this.deps.errors.record({ source: "schedule", diff --git a/packages/server/src/services/project-config-service.ts b/packages/server/src/services/project-config-service.ts index b22ea80..b6e31ff 100644 --- a/packages/server/src/services/project-config-service.ts +++ b/packages/server/src/services/project-config-service.ts @@ -14,6 +14,14 @@ * sent to AgentHub verbatim — string concatenation like `/` is * forbidden everywhere in the pipeline. `default_model` / `vision_model` are `{ * provider, model_id }` paired references (TOML tables). + * + * Reads are memoized on the file's mtime (see readTable): every consumer of this + * service — scheduler model-ref validation, the models/schedules routes, usage + * pricing — shares one parsed table per on-disk version instead of re-reading and + * re-parsing the TOML per call. The service's own writes all funnel through + * `writeRaw`, which invalidates synchronously; external edits (CLI, hand edits) are + * caught by the stat. The cached table is shared between callers and must be treated + * as immutable — every mutating method here copies before changing. */ import fs from "node:fs/promises"; import path from "node:path"; @@ -24,12 +32,13 @@ import { GenerativeModel, catalogEntryFor, defaultProjectConfig, + projectConfigFromTable, projectConfigPath, renderProjectConfigToml, resolveModelEnv, userText, } from "@prismshadow/penguin-core"; -import type { LLMOutcome, ModelRef, OmniMessage } from "@prismshadow/penguin-core"; +import type { LLMOutcome, ModelRef, OmniMessage, ProjectConfig } from "@prismshadow/penguin-core"; import type { ChatDefaultsDto, ModelInfo, @@ -41,6 +50,7 @@ import type { ModelTestResponse, } from "../api/types.js"; import { badRequest } from "../http/validate.js"; +import { cacheable } from "../internal/mtime-gate.js"; import type { PricingRates } from "./usage-service.js"; type RawTable = Record; @@ -118,6 +128,15 @@ const SPEED_PROBE_PROMPT = "Count from 1 to 50 as a comma-separated list, and nothing else. Do not think or explain.\n"; export class ProjectConfigService { + /** + * Parsed-table cache, one entry per Project, keyed by the config file's mtime as + * recorded at read time (a fresh mtime is stored as a never-matching sentinel, see + * mtime-gate). A repeat read while the stat still matches costs one stat and zero + * parses; a mismatch (external edit) or a service write (writeRaw deletes the + * entry) falls back to a full read. + */ + private readonly cache = new Map(); + constructor(private readonly root: string) {} private filePath(projectId: string): string { @@ -140,14 +159,50 @@ export class ProjectConfigService { /** Reads the raw TOML object; returns an empty object if the file doesn't exist (does not write to disk). */ async readRaw(projectId: string): Promise { + return (await this.readTable(projectId)) ?? {}; + } + + /** + * mtime-gated read of the parsed table; null when the file doesn't exist (readRaw + * flattens that to `{}`, loadConfig to the preset default config — the two + * pre-existing missing-file behaviors). Serving the shared cached object is safe + * because no consumer mutates it (see the class header). + */ + private async readTable(projectId: string): Promise { + const file = this.filePath(projectId); + let mtimeMs: number; + try { + mtimeMs = (await fs.stat(file)).mtimeMs; + } catch (err) { + if ((err as NodeJS.ErrnoException).code !== "ENOENT") throw err; + this.cache.delete(projectId); // Config deleted (or the whole Project): drop the stale entry. + return null; + } + const cached = this.cache.get(projectId); + if (cached && cached.mtimeMs === mtimeMs) return cached.table; let raw: string; try { - raw = await fs.readFile(this.filePath(projectId), "utf8"); + raw = await fs.readFile(file, "utf8"); } catch (err) { - if ((err as NodeJS.ErrnoException).code === "ENOENT") return {}; - throw err; + if ((err as NodeJS.ErrnoException).code !== "ENOENT") throw err; + this.cache.delete(projectId); // Vanished between stat and read: same as never existing. + return null; } - return asTable(parseToml(raw)); + const table = asTable(parseToml(raw)); + this.cache.set(projectId, { mtimeMs: cacheable(mtimeMs), table }); + return table; + } + + /** + * Typed view of the same cached table — byte-for-byte the semantics of core's + * `loadProjectConfig` (missing file → preset default config; legacy format → the + * same error, via core's shared narrowing) without its per-call readFile + parse. + * Serves the scheduler's model-ref validation (see schedule-store). + */ + async loadConfig(projectId: string): Promise { + const table = await this.readTable(projectId); + if (table === null) return defaultProjectConfig(); + return projectConfigFromTable(this.filePath(projectId), table); } /** @@ -164,6 +219,9 @@ export class ProjectConfigService { // (the same file should never have two formats). await fs.writeFile(file, renderProjectConfigToml(data), { encoding: "utf8", mode: 0o600 }); await fs.chmod(file, 0o600); + // Every service write funnels through here: invalidate synchronously so the next + // read re-parses (external writers are caught by readTable's stat instead). + this.cache.delete(projectId); } /** diff --git a/packages/server/src/services/trace-index.ts b/packages/server/src/services/trace-index.ts index 8e05daf..ca525e2 100644 --- a/packages/server/src/services/trace-index.ts +++ b/packages/server/src/services/trace-index.ts @@ -34,6 +34,7 @@ import { agentsDir, isSessionMeta, tracesDir } from "@prismshadow/penguin-core"; import type { OmniMessage } from "@prismshadow/penguin-core"; import type { TraceFileRow, TraceSessionRow } from "../db/repos/trace-index.js"; import { TraceIndexRepo } from "../db/repos/trace-index.js"; +import { cacheable, statMtime } from "../internal/mtime-gate.js"; import { readTraceHead } from "../internal/trace-head.js"; import { asSessionSource } from "../runtime/session-sources.js"; import type { SessionSources } from "../runtime/session-sources.js"; @@ -41,35 +42,12 @@ import { fallbackTitle } from "../runtime/title-generator.js"; const TRACE_FILE_RE = /^(.+)_(\d{3})\.jsonl$/; -/** - * mtimes younger than this are treated as UNSTABLE and never cached as clean: - * filesystem timestamps are coarse (a write and a later change can land on the same - * tick), so a fresh directory keeps being re-diffed until it has been quiet for this - * long — an active Agent costs one small readdir per request while writing, and the - * gate can never wedge on a same-tick change. - */ -const FRESH_MS = 2000; - -/** A gate-cacheable mtime: the real value once stable, else a sentinel that never matches. */ -function cacheable(mtimeMs: number): number { - return Date.now() - mtimeMs > FRESH_MS ? mtimeMs : -1; -} - /** Absolute path of an indexed shard (reconstructed — rows never store paths; the data root may move). */ export function traceFilePath(root: string, row: TraceFileRow): string { const name = `${row.sessionId}_${String(row.fileIndex).padStart(3, "0")}.jsonl`; return path.join(tracesDir(root, row.projectId, row.agentId), row.date, name); } -/** mtimeMs of a path; null when it does not exist. */ -async function statMtime(p: string): Promise { - try { - return (await fs.stat(p)).mtimeMs; - } catch { - return null; - } -} - async function listDirs(dir: string): Promise { try { const entries = await fs.readdir(dir, { withFileTypes: true }); @@ -192,7 +170,7 @@ export class TraceIndexService { // a new file inside an EXISTING date dir — in practice always the newest one (the // Writer names dirs by current local date) — so gate on that single dir's mtime too. // Cached sentinels (-1: the dir was fresh when last seen) never match, forcing a - // re-diff until the tree has been quiet (see FRESH_MS). + // re-diff until the tree has been quiet (see mtime-gate's FRESH_MS). const newest = [...cached.dates.keys()].sort().at(-1); if (newest === undefined) return; this.counters.gateStats += 1; diff --git a/packages/server/test/project-config-cache.test.ts b/packages/server/test/project-config-cache.test.ts new file mode 100644 index 0000000..3a5f5f3 --- /dev/null +++ b/packages/server/test/project-config-cache.test.ts @@ -0,0 +1,106 @@ +/** + * ProjectConfigService read-cache semantics: repeat reads are served from the + * mtime-keyed parsed-table cache (one initial readFile), the service's own writes + * invalidate synchronously, an external edit is caught by the stat, a fresh mtime is + * never cached as clean (same-second-granularity guard), and the typed loadConfig + * view mirrors core's loadProjectConfig (missing file → the preset default config). + */ +import fs from "node:fs/promises"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + defaultProjectConfig, + projectConfigPath, + renderProjectConfigToml, +} from "@prismshadow/penguin-core"; +import { ProjectConfigService } from "../src/services/project-config-service.js"; +import { makeTempRoot } from "./helpers.js"; + +const P = "project-cfg"; +const PRICED = { + provider: "custom", + model_id: "m1", + pricing: { unit: "usd_per_mtok", cache_read: 1, cache_write: 2, output: 3 }, +}; + +/** Ages the config's mtime past the gate's FRESH_MS window (fixed instants keep backdates idempotent). */ +async function backdate(file: string, instant: string): Promise { + const at = new Date(instant); + await fs.utimes(file, at, at); +} + +describe("project-config read cache", () => { + let root: string; + let svc: ProjectConfigService; + let file: string; + + beforeEach(async () => { + root = await makeTempRoot(); + svc = new ProjectConfigService(root); + file = projectConfigPath(root, P); + await svc.writeRaw(P, { name: "cached", models: [PRICED] }); + await backdate(file, "2026-01-01T00:00:00Z"); + }); + afterEach(async () => { + vi.restoreAllMocks(); + await fs.rm(root, { recursive: true, force: true }); + }); + + it("repeat reads cost one readFile: pricing, name and the typed loadConfig view share the cached parse", async () => { + const reads = vi.spyOn(fs, "readFile"); + expect(await svc.getPricing(P, "custom", "m1")).toEqual({ + cacheRead: 1, + cacheWrite: 2, + output: 3, + }); + expect(await svc.getPricing(P, "custom", "m1")).toEqual({ + cacheRead: 1, + cacheWrite: 2, + output: 3, + }); + expect(await svc.getName(P)).toBe("cached"); + expect((await svc.loadConfig(P)).models.map((m) => m.model_id)).toEqual(["m1"]); + expect(reads).toHaveBeenCalledTimes(1); + }); + + it("a service write invalidates synchronously; once quiet, the next read re-parses exactly once", async () => { + await svc.getName(P); // Warm the cache at the stable mtime. + await svc.setName(P, "renamed"); + await backdate(file, "2026-01-01T00:01:00Z"); + const reads = vi.spyOn(fs, "readFile"); + expect(await svc.getName(P)).toBe("renamed"); + expect(await svc.getPricing(P, "custom", "m1")).toEqual({ + cacheRead: 1, + cacheWrite: 2, + output: 3, + }); + expect(reads).toHaveBeenCalledTimes(1); + }); + + it("an external edit (bypassing the service) is caught by the stat: new mtime → re-read", async () => { + expect(await svc.getName(P)).toBe("cached"); // Warm at the old mtime. + await fs.writeFile( + file, + renderProjectConfigToml({ name: "edited", models: [{ provider: "custom", model_id: "m2" }] }), + "utf8", + ); + await backdate(file, "2026-01-01T00:02:00Z"); // Stable, but a different mtime than cached. + expect(await svc.getName(P)).toBe("edited"); + expect((await svc.loadConfig(P)).models.map((m) => m.model_id)).toEqual(["m2"]); + expect(await svc.getPricing(P, "custom", "m1")).toBeUndefined(); + }); + + it("a fresh mtime is never cached as clean: reads keep hitting disk until the file has been quiet", async () => { + await svc.setName(P, "fresh"); // mtime = now, inside the FRESH_MS window. + const reads = vi.spyOn(fs, "readFile"); + await svc.getName(P); + await svc.getName(P); + expect(reads).toHaveBeenCalledTimes(2); + }); + + it("missing file: readRaw → {} and loadConfig → the preset default config (core loadProjectConfig parity)", async () => { + await svc.getName(P); // Warm, then delete out from under the cache. + await fs.rm(file); + expect(await svc.readRaw(P)).toEqual({}); + expect(await svc.loadConfig(P)).toEqual(defaultProjectConfig()); + }); +}); diff --git a/packages/server/test/scheduler.test.ts b/packages/server/test/scheduler.test.ts index 3d4ec88..9d4e68e 100644 --- a/packages/server/test/scheduler.test.ts +++ b/packages/server/test/scheduler.test.ts @@ -9,8 +9,13 @@ */ import fs from "node:fs/promises"; import path from "node:path"; -import { afterEach, beforeEach, describe, expect, it } from "vitest"; -import { saveProjectConfig, scheduleDir } from "@prismshadow/penguin-core"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + agentsDir, + projectConfigPath, + saveProjectConfig, + scheduleDir, +} from "@prismshadow/penguin-core"; import { openDatabase } from "../src/db/database.js"; import { ProjectsRepo } from "../src/db/repos/projects.js"; import { SchedulesRepo } from "../src/db/repos/schedules.js"; @@ -18,6 +23,7 @@ import { SessionsRepo } from "../src/db/repos/sessions.js"; import { UsersRepo } from "../src/db/repos/users.js"; import type { ErrorRecordArgs } from "../src/runtime/error-recorder.js"; import { Scheduler } from "../src/runtime/scheduler.js"; +import { ProjectConfigService } from "../src/services/project-config-service.js"; import type { ScheduleServerEvent } from "../src/api/types.js"; import { makeTempRoot } from "./helpers.js"; @@ -95,6 +101,7 @@ describe("scheduler", () => { return { sessionId }; }, }, + projectConfig: new ProjectConfigService(root), errors: { record: (args) => void errors.push(args) }, notify: (userId, event) => void events.push({ userId, event }), now: () => nowMs, @@ -132,6 +139,104 @@ describe("scheduler", () => { return new Date(ms).toISOString(); } + /** + * Ages the given paths' mtimes past the mtime gate's FRESH_MS window (fresh mtimes are + * deliberately never cached as clean), so steady-state assertions see a tree that looks + * quiet — exactly like files written minutes ago. Fixed instants keep backdates idempotent + * and let a later edit be given a *different* stable mtime. + */ + async function backdate(paths: string[], instant = "2026-01-01T00:00:00Z"): Promise { + const at = new Date(instant); + for (const p of paths) await fs.utimes(p, at, at); + } + + it("steady state: an unchanged tree ticks with zero file reads and zero dir listings (stat-gated); a new file re-scans", async () => { + insertSession("session-1"); + // Carries a model reference so every tick also exercises model-ref validation — + // the per-tick .project_config.toml read+parse this change eliminates. + await writeFile("cached", [ + `prompt = "cached prompt"`, + `enabled = true`, + `start_at = "${iso(T0 + 60 * MIN)}"`, + `period = "30m"`, + `provider = "custom"`, + `model_id = "m-bench"`, + ]); + await backdate([ + agentsDir(root, P), + scheduleDir(root, P, A), + path.join(scheduleDir(root, P, A), "cached.toml"), + projectConfigPath(root, P), + ]); + await scheduler.tickOnce(); // First look: populates the gate caches (this pass reads). + expect(repo.find(P, A, "cached")).not.toBeNull(); + + const fileReads = scheduler.files.counters.fileReads; + const dirScans = scheduler.files.counters.dirScans; + const readFileSpy = vi.spyOn(fs, "readFile"); + const readdirSpy = vi.spyOn(fs, "readdir"); + await scheduler.tickOnce(); + await scheduler.tickOnce(); + // Unchanged tree: the ticks are pure stats — no schedule-file reads, no readdir of the + // agents/schedule dirs, and no .project_config.toml read for model-ref validation either. + expect(scheduler.files.counters.fileReads).toBe(fileReads); + expect(scheduler.files.counters.dirScans).toBe(dirScans); + expect(readFileSpy).not.toHaveBeenCalled(); + expect(readdirSpy).not.toHaveBeenCalled(); + readFileSpy.mockRestore(); + readdirSpy.mockRestore(); + + // A new file moves the schedule dir's mtime: the gate notices and registers it. + await writeFile("added", [ + `prompt = "a"`, + `enabled = true`, + `start_at = "${iso(T0 + 60 * MIN)}"`, + `session_id = "session-1"`, + ]); + await scheduler.tickOnce(); + expect(repo.find(P, A, "added")).not.toBeNull(); + expect(scheduler.files.counters.dirScans).toBe(dirScans + 1); + }); + + it("an in-place edit (file mtime moves, dir mtime does not) is picked up next tick and fires with the new content", async () => { + insertSession("session-1"); + await writeFile("inplace", [ + `prompt = "old prompt"`, + `enabled = true`, + `start_at = "${iso(T0 + 5 * MIN)}"`, + `period = "5m"`, + `session_id = "session-1"`, + ]); + await backdate([ + agentsDir(root, P), + scheduleDir(root, P, A), + path.join(scheduleDir(root, P, A), "inplace.toml"), + projectConfigPath(root, P), + ]); + await scheduler.tickOnce(); // Populate the gate caches at the stable mtimes. + await scheduler.tickOnce(); // Steady state: served from memory. + + // Overwriting an existing file's CONTENT never moves the directory's mtime — only the + // file's own — which is exactly what the per-file stats of the unchanged-dir path catch. + await writeFile("inplace", [ + `prompt = "new prompt"`, + `enabled = true`, + `start_at = "${iso(T0 + 5 * MIN)}"`, + `period = "5m"`, + `session_id = "session-1"`, + ]); + await backdate([path.join(scheduleDir(root, P, A), "inplace.toml")], "2026-01-01T00:01:00Z"); + + nowMs = T0 + 5 * MIN; + const dirScans = scheduler.files.counters.dirScans; + await scheduler.tickOnce(); + expect(started).toHaveLength(1); + expect(started[0]?.text).toContain("new prompt"); + expect(started[0]?.text).not.toContain("old prompt"); + // The pickup came from the unchanged-dir path's per-file stat, not a directory rescan. + expect(scheduler.files.counters.dirScans).toBe(dirScans); + }); + it("periodic task: registration consumes past slots (missed, not backfilled); fires once on time, never twice", async () => { insertSession("session-1"); await writeFile("report", [ diff --git a/packages/server/test/schedules.test.ts b/packages/server/test/schedules.test.ts index 3d9feaa..0af68d0 100644 --- a/packages/server/test/schedules.test.ts +++ b/packages/server/test/schedules.test.ts @@ -183,6 +183,37 @@ describe("schedules api", () => { expect(stale?.creatorUserId).toBe("owner_a"); }); + it("a hand-edited file is reconciled past a warm mtime cache (an in-place edit moves only the file's mtime)", async () => { + expect( + ( + await owner.post(base, { + name: "hand", + prompt: "before", + enabled: true, + startAt: FUTURE, + period: "30m", + }) + ).status, + ).toBe(201); + // Make the tree look quiet (a fresh mtime is never cached as clean), then warm the + // scan cache with one GET — the steady state the schedule routes now serve from. + const dir = scheduleDir(t.root, projectId, "default_agent"); + const file = path.join(dir, "hand.toml"); + const old = new Date("2026-01-01T00:00:00Z"); + await fs.utimes(dir, old, old); + await fs.utimes(file, old, old); + let list = (await (await owner.get(base)).json()) as SchedulesResponse; + expect(list.schedules[0]?.prompt).toBe("before"); + // Hand-edit the file in place (like over SSH): the content and file mtime change, + // the directory's mtime does NOT — the per-file stat still catches it. + const raw = await fs.readFile(file, "utf8"); + await fs.writeFile(file, raw.replace('"before"', '"after"'), "utf8"); + const edited = new Date("2026-01-01T00:01:00Z"); + await fs.utimes(file, edited, edited); + list = (await (await owner.get(base)).json()) as SchedulesResponse; + expect(list.schedules[0]?.prompt).toBe("after"); + }); + it("a file holding model_id without provider lands in invalidFiles instead of being scheduled", async () => { // What a schedule file persisted before the pairing rule looks like: the provider is // never filled in for it, the file is simply reported invalid and skipped.