Compare commits

...
Author SHA1 Message Date
Mohamed Boudra 0290b0e35b fix(server): complete daemon teardown after failures 2026-08-13 22:46:09 +02:00
Mohamed Boudra 07ffe243fb fix(nix): update npm dependency hash 2026-08-13 22:32:44 +02:00
Mohamed Boudra ae3c0119f9 fix(storage): close database after bootstrap failure
The bootstrap wrapper releases the SQLite handle when daemon construction fails before returning the normal shutdown owner.
2026-08-13 22:24:14 +02:00
Mohamed Boudra 5d195e7e15 feat(storage): add SQLite persistence foundation
Begin the storage migration with a bootstrap-owned database and one-way legacy imports for schedules and push tokens. Legacy JSON remains untouched for downgrade compatibility while SQLite becomes the sole authority after first open.
2026-08-13 22:24:14 +02:00
26 changed files with 1057 additions and 274 deletions
+31 -10
View File
@@ -30,7 +30,20 @@ Workspace archive runs lifecycle teardown from the exact `cwd` but removes only
`worktreeRoot` after its last active reference disappears. Worktree recovery recreates that backing
checkout from `mainRepoRoot`, then restores the relative path from `worktreeRoot` to `cwd`.
Paseo uses **file-based JSON persistence** instead of a traditional database. All data is validated at runtime with Zod schemas. Most stores write atomically (write to temp file, then rename); a few still use plain `writeFile` — see each section. There is no schema-versioning/migration framework — schemas rely on optional fields with defaults for forward compatibility, with a small amount of inline normalization in `persisted-config.ts` for legacy provider/speech entries.
Paseo stores schedules and push tokens in `$PASEO_HOME/paseo.db`. The remaining record stores use
JSON files. SQLite migrations are ordered SQL scripts tracked by `PRAGMA user_version`. Record
tables keep identity and queried fields in columns and the complete record in a JSON `payload`
column. Reads validate that payload with the same Zod schema used at the store boundary.
Each SQLite-backed store owns a one-way import from its legacy JSON path. On first open, an empty
table imports valid legacy records in one transaction and writes a marker to `legacy_imports`.
Invalid records are skipped and logged. Once the marker exists, SQLite is the only authority for
that store: the daemon does not read from or write to the legacy path. Legacy files remain untouched
so downgrading resumes from the state captured when the import ran; there is no automatic
down-migration.
JSON stores rely on optional fields with defaults for forward compatibility. Most write atomically
with a temporary file and rename; see each section for exceptions.
All server-side stores live under `$PASEO_HOME` (defaults to `~/.paseo`).
@@ -49,11 +62,12 @@ $PASEO_HOME/
├── daemon-keypair.json # E2EE keypair for relay (mode 0600)
├── paseo.pid # Daemon PID lock file
├── daemon.log # Default log file (path configurable)
├── paseo.db # SQLite schedules, push tokens, and import markers
├── agents/
│ └── {sanitized-cwd}/
│ └── {agentId}.json # One file per agent
├── schedules/
│ └── {scheduleId}.json # One file per schedule
├── schedules/ # Legacy schedule import source, left untouched
│ └── {scheduleId}.json
├── projects/
│ ├── projects.json # Project registry
│ ├── workspaces.json # Workspace registry
@@ -61,10 +75,10 @@ $PASEO_HOME/
├── runtime/
│ └── managed-processes/
│ └── {recordId}.json # Helper processes owned by Paseo; reconciled on daemon bootstrap
└── push-tokens.json # Expo push notification tokens
└── push-tokens.json # Legacy push token import source, left untouched
```
The `agents/{sanitized-cwd}/` directory name is derived from the agent's `cwd` by stripping the filesystem root and replacing path separators with `-` (Windows drive letters become a `C-` style prefix). Persistent server stores write atomically by writing a temp file in the target directory and then renaming it into place.
The `agents/{sanitized-cwd}/` directory name is derived from the agent's `cwd` by stripping the filesystem root and replacing path separators with `-` (Windows drive letters become a `C-` style prefix). File-backed record stores write atomically by writing a temp file in the target directory and then renaming it into place.
---
@@ -325,9 +339,11 @@ Paseo uses these paths under the configured OpenAI base URL:
## 3. Schedule
**Path:** `$PASEO_HOME/schedules/{id}.json`
**Table:** `paseo.db:schedules`
One file per schedule. ID is 8 hex characters.
Each row stores `id`, `created_at`, the active name-and-target `identity_key`, and the validated
schedule JSON in `payload`. ID is 8 hex characters. The legacy
`$PASEO_HOME/schedules/{id}.json` files are import sources only.
| Field | Type | Description |
| ----------- | ------------------------------------- | -------------------------------- |
@@ -442,15 +458,20 @@ than treating it as valid.
## 6. Push Token Store
**Path:** `$PASEO_HOME/push-tokens.json`
**Table:** `paseo.db:push_tokens`
Legacy import source:
```json
{
"tokens": ["ExponentPushToken[...]", ...]
"subscriptions": [{ "token": "ExponentPushToken[...]", "expiresAt": "2026-08-12T00:00:00.000Z" }]
}
```
Simple set of Expo push notification tokens. Loaded with permissive parsing (filters non-string entries). Persisted with atomic temp-file rename.
The older `{ "tokens": [...] }` shape is imported with a fresh lease. Each token is a row keyed by
its normalized string, with the validated subscription record in `payload`.
`$PASEO_HOME/push-tokens.json` is read once as an import source. Non-string and empty legacy entries
are skipped and logged.
---
+1 -1
View File
@@ -1 +1 @@
sha256-i5PbVUe2Ec+GtghV9IpCJQJ9hcUT5hFhmxneNvoD584=
sha256-BeS2/PYAPoG4U3q7N2qOVuPC1T5sDQ+c9hv2cTn3KRU=
+17 -1
View File
@@ -37163,6 +37163,9 @@
"typescript": "^5.2.2",
"vitest": "^4.1.6",
"zx": "^8.8.5"
},
"engines": {
"node": ">=22.20.0"
}
},
"packages/cli/node_modules/@vitest/expect": {
@@ -38845,7 +38848,7 @@
"devDependencies": {
"@playwright/test": "^1.56.1",
"@types/express": "^4.17.20",
"@types/node": "^20.9.0",
"@types/node": "^22.20.1",
"@types/qrcode": "^1.5.6",
"@types/uuid": "^9.0.7",
"@types/ws": "^8.5.8",
@@ -38854,6 +38857,9 @@
"tsx": "^4.6.0",
"typescript": "^5.2.2",
"vitest": "^4.1.6"
},
"engines": {
"node": ">=22.20.0"
}
},
"packages/server/node_modules/@anthropic-ai/claude-agent-sdk": {
@@ -39002,6 +39008,16 @@
"cross-spawn": "7.0.6"
}
},
"packages/server/node_modules/@types/node": {
"version": "22.20.1",
"resolved": "https://registry.npmjs.org/@types/node/-/node-22.20.1.tgz",
"integrity": "sha512-EANqOCF9QFyra+4pfxUcX9STKJpCLjMbObVzljIJomAWSnuSIEAvyzEU53GaajbXJEgdh0iEcPL+DGvpUd4k1Q==",
"dev": true,
"license": "MIT",
"dependencies": {
"undici-types": "~6.21.0"
}
},
"packages/server/node_modules/@vitest/expect": {
"version": "4.1.6",
"resolved": "https://registry.npmjs.org/@vitest/expect/-/expect-4.1.6.tgz",
+3
View File
@@ -46,5 +46,8 @@
"typescript": "^5.2.2",
"vitest": "^4.1.6",
"zx": "^8.8.5"
},
"engines": {
"node": ">=22.20.0"
}
}
+5 -1
View File
@@ -51,6 +51,7 @@
"speech:models": "tsx scripts/list-speech-models.ts",
"speech:download": "tsx scripts/download-speech-models.ts",
"speech:transcribe:local": "tsx scripts/transcribe-local-wav.ts",
"db:query": "../../scripts/dev-home.sh tsx scripts/db-query.ts",
"test": "npm run test:unit && npm run test:integration",
"test:unit": "vitest run --fileParallelism --exclude \"**/*.e2e.test.ts\"",
"test:integration": "vitest run --maxWorkers=1 src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts && vitest run --maxWorkers=1 src/server/session.create-agent-worktree-autoarchive.e2e.test.ts -t \"creates a worktree and auto-archives both\"",
@@ -107,7 +108,7 @@
"devDependencies": {
"@playwright/test": "^1.56.1",
"@types/express": "^4.17.20",
"@types/node": "^20.9.0",
"@types/node": "^22.20.1",
"@types/qrcode": "^1.5.6",
"@types/uuid": "^9.0.7",
"@types/ws": "^8.5.8",
@@ -119,5 +120,8 @@
},
"overrides": {
"lightningcss": "1.30.1"
},
"engines": {
"node": ">=22.20.0"
}
}
+20
View File
@@ -0,0 +1,20 @@
import { resolvePaseoHome } from "../src/server/paseo-home.js";
import { openDatabase } from "../src/server/db/open.js";
const sql = process.argv.slice(2).join(" ").trim();
if (!sql) {
throw new Error('Usage: npm run db:query -- "SELECT ..."');
}
const database = openDatabase(resolvePaseoHome());
try {
const statement = database.prepare(sql);
if (statement.columns().length > 0) {
console.table(statement.all());
} else {
const result = statement.run();
console.log(`changes: ${result.changes}`);
}
} finally {
database.close();
}
@@ -0,0 +1,41 @@
import { describe, expect, test } from "vitest";
import { runShutdownSequence } from "./bootstrap.js";
describe("runShutdownSequence", () => {
test("runs every step before rethrowing the first failure", async () => {
const firstFailure = new Error("first failure");
const secondFailure = new Error("second failure");
const completed: string[] = [];
const reported: Array<{ name: string; error: unknown }> = [];
const shutdown = runShutdownSequence(
[
["first", () => completed.push("first")],
[
"second",
() => {
completed.push("second");
throw firstFailure;
},
],
[
"third",
async () => {
completed.push("third");
throw secondFailure;
},
],
["fourth", () => completed.push("fourth")],
],
(name, error) => reported.push({ name, error }),
);
await expect(shutdown).rejects.toBe(firstFailure);
expect(completed).toEqual(["first", "second", "third", "fourth"]);
expect(reported).toEqual([
{ name: "second", error: firstFailure },
{ name: "third", error: secondFailure },
]);
});
});
+90 -34
View File
@@ -79,6 +79,32 @@ export function parseListenString(listen: string): ListenTarget {
throw new Error(`Invalid listen string: ${listen}`);
}
type ShutdownStep = readonly [name: string, run: () => void | Promise<void>];
export async function runShutdownSequence(
steps: readonly ShutdownStep[],
onFailure: (name: string, error: unknown) => void,
): Promise<void> {
let failed = false;
let firstFailure: unknown;
for (const [name, run] of steps) {
try {
await run();
} catch (error) {
if (!failed) {
failed = true;
firstFailure = error;
}
onFailure(name, error);
}
}
if (failed) {
throw firstFailure;
}
}
function formatListenTarget(listenTarget: ListenTarget | null): string | null {
if (!listenTarget) {
return null;
@@ -216,6 +242,8 @@ import {
type HubRelationshipRemote,
} from "./hub/relationship-remote.js";
import { DaemonExecutions } from "./hub/daemon-executions.js";
import { openDatabase } from "./db/open.js";
import { PushTokenStore } from "./push/token-store.js";
const MAX_MCP_DEBUG_BATCH_ITEMS = 10;
const REDACTED_LOG_VALUE = "[redacted]";
@@ -552,6 +580,21 @@ export async function createPaseoDaemon(
): Promise<PaseoDaemon> {
configureGitProcessPolicy(config.git ?? resolveGitProcessPolicy({ env: process.env }));
const logger = rootLogger.child({ module: "bootstrap" });
const database = openDatabase(config.paseoHome);
try {
return await createPaseoDaemonWithDatabase(config, logger, dependencies, database);
} catch (error) {
database.close();
throw error;
}
}
async function createPaseoDaemonWithDatabase(
config: PaseoDaemonConfig,
logger: Logger,
dependencies: PaseoDaemonDependencies,
database: ReturnType<typeof openDatabase>,
): Promise<PaseoDaemon> {
const bootstrapStart = performance.now();
const elapsed = () => `${(performance.now() - bootstrapStart).toFixed(0)}ms`;
const daemonVersion = config.daemonVersion ?? resolveDaemonVersion(import.meta.url);
@@ -1205,6 +1248,7 @@ export async function createPaseoDaemon(
};
const scheduleService = new ScheduleService({
paseoHome: config.paseoHome,
database,
logger,
agentManager,
agentStorage,
@@ -1222,6 +1266,11 @@ export async function createPaseoDaemon(
}
});
logger.info({ elapsed: elapsed() }, "Schedule service initialized");
const pushTokenStore = new PushTokenStore({
database,
legacyFilePath: path.join(config.paseoHome, "push-tokens.json"),
logger,
});
logger.info({ elapsed: elapsed() }, "Loading persisted agent registry");
const persistedRecords = await agentStorage.list();
logger.info(
@@ -1504,6 +1553,7 @@ export async function createPaseoDaemon(
agentManager,
agentStorage,
downloadTokenStore,
pushTokenStore,
config.paseoHome,
daemonConfigStore,
mcpBaseUrl,
@@ -1613,40 +1663,46 @@ export async function createPaseoDaemon(
};
const stop = async () => {
await hubRelationships.stop();
workspaceReconciliation.dispose();
scriptHealthMonitor.stop();
// Freeze both ingress and registration before taking the agent closure snapshot.
wsServer?.prepareForShutdown();
agentManager.prepareForShutdown();
await closeAllAgents(logger, agentManager);
await agentManager.flushForShutdown().catch(() => undefined);
detachAgentStoragePersistence();
await agentStorage.flush().catch(() => undefined);
await providerSnapshotManager.shutdown();
terminalManager.killAll();
speechService.stop();
await scheduleService.stop().catch(() => undefined);
await relayRuntime?.stop().catch(() => undefined);
if (wsServer) {
await wsServer.close();
}
await serviceProxy.stopStandalone();
// Force-drop remaining sockets so httpServer.close() resolves promptly.
// We've already closed wsServer (which sent ws-layer close frames) and
// stopped every other service, so anything still attached is a TCP
// socket whose higher-level shutdown hasn't fully released it (e.g.
// upgraded WS sockets in the closing handshake, or HTTP keep-alive
// sockets in CLOSE_WAIT). closeIdleConnections() does not catch
// upgraded sockets, so we use closeAllConnections() here.
httpServer.closeAllConnections();
await new Promise<void>((resolve) => {
httpServer.close(() => resolve());
});
// Clean up socket files
if (listenTarget.type === "socket" && existsSync(listenTarget.path)) {
unlinkSync(listenTarget.path);
}
await runShutdownSequence(
[
["hub relationships", () => hubRelationships.stop()],
["workspace reconciliation", () => workspaceReconciliation.dispose()],
["script health monitor", () => scriptHealthMonitor.stop()],
// Freeze both ingress and registration before taking the agent closure snapshot.
["WebSocket ingress", () => wsServer?.prepareForShutdown()],
["agent ingress", () => agentManager.prepareForShutdown()],
["agents", () => closeAllAgents(logger, agentManager)],
["agent manager persistence", () => agentManager.flushForShutdown()],
["agent storage persistence hook", () => detachAgentStoragePersistence()],
["agent storage", () => agentStorage.flush()],
["provider snapshots", () => providerSnapshotManager.shutdown()],
["terminals", () => terminalManager.killAll()],
["speech", () => speechService.stop()],
["schedules", () => scheduleService.stop()],
["relay", () => relayRuntime?.stop()],
["WebSocket server", () => wsServer?.close()],
["service proxy", () => serviceProxy.stopStandalone()],
// Force-drop remaining sockets so httpServer.close() resolves promptly.
// We've already closed wsServer (which sent ws-layer close frames) and
// stopped every other service, so anything still attached is a TCP
// socket whose higher-level shutdown hasn't fully released it (e.g.
// upgraded WS sockets in the closing handshake, or HTTP keep-alive
// sockets in CLOSE_WAIT). closeIdleConnections() does not catch
// upgraded sockets, so we use closeAllConnections() here.
["HTTP connections", () => httpServer.closeAllConnections()],
["HTTP server", () => new Promise<void>((resolve) => httpServer.close(() => resolve()))],
[
"socket file",
() => {
if (listenTarget.type === "socket" && existsSync(listenTarget.path)) {
unlinkSync(listenTarget.path);
}
},
],
["database", () => database.close()],
],
(name, error) => logger.warn({ err: error, step: name }, "Daemon shutdown step failed"),
);
};
return {
@@ -0,0 +1,26 @@
import type { DatabaseSync } from "node:sqlite";
export interface LegacyImportResult {
importedCount: number;
skippedCount: number;
}
export function hasLegacyImportMarker(database: DatabaseSync, store: string): boolean {
const row = database.prepare("SELECT 1 FROM legacy_imports WHERE store = ?").get(store) as
| { 1: number }
| undefined;
return row !== undefined;
}
export function recordLegacyImportMarker(
database: DatabaseSync,
store: string,
result: LegacyImportResult,
): void {
database
.prepare(
`INSERT INTO legacy_imports (store, imported_at, imported_count, skipped_count)
VALUES (?, ?, ?, ?)`,
)
.run(store, new Date().toISOString(), result.importedCount, result.skippedCount);
}
@@ -0,0 +1,69 @@
import type { DatabaseSync } from "node:sqlite";
import { runInTransaction } from "./transaction.js";
interface Migration {
version: number;
sql: string;
}
const MIGRATIONS: readonly Migration[] = [
{
version: 1,
sql: `
CREATE TABLE legacy_imports (
store TEXT PRIMARY KEY,
imported_at TEXT NOT NULL,
imported_count INTEGER NOT NULL,
skipped_count INTEGER NOT NULL
) STRICT;
`,
},
{
version: 2,
sql: `
CREATE TABLE schedules (
id TEXT PRIMARY KEY,
created_at TEXT NOT NULL,
identity_key TEXT,
payload TEXT NOT NULL
) STRICT;
CREATE INDEX schedules_created_at_idx ON schedules(created_at);
CREATE INDEX schedules_identity_key_idx ON schedules(identity_key);
`,
},
{
version: 3,
sql: `
CREATE TABLE push_tokens (
token TEXT PRIMARY KEY,
payload TEXT NOT NULL
) STRICT;
`,
},
];
function readUserVersion(database: DatabaseSync): number {
const row = database.prepare("PRAGMA user_version").get() as { user_version: number };
return row.user_version;
}
export function runMigrations(database: DatabaseSync): void {
const currentVersion = readUserVersion(database);
const latestVersion = MIGRATIONS.at(-1)?.version ?? 0;
if (currentVersion > latestVersion) {
throw new Error(
`paseo.db schema version ${currentVersion} is newer than this daemon supports (${latestVersion})`,
);
}
for (const migration of MIGRATIONS) {
if (migration.version <= currentVersion) {
continue;
}
runInTransaction(database, () => {
database.exec(migration.sql);
database.exec(`PRAGMA user_version = ${migration.version}`);
});
}
}
@@ -0,0 +1,72 @@
import { existsSync } from "node:fs";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, describe, expect, test } from "vitest";
import { assertSupportedNodeVersion, openDatabase } from "./open.js";
import { runInTransaction } from "./transaction.js";
const homes: string[] = [];
afterEach(async () => {
await Promise.all(homes.splice(0).map((home) => rm(home, { recursive: true, force: true })));
});
async function createHome(): Promise<string> {
const home = await mkdtemp(join(tmpdir(), "paseo-db-test-"));
homes.push(home);
return home;
}
function insertThenFail(database: ReturnType<typeof openDatabase>): void {
runInTransaction(database, () => {
database
.prepare("INSERT INTO push_tokens (token, payload) VALUES (?, ?)")
.run("token", JSON.stringify({ token: "token" }));
throw new Error("stop");
});
}
describe("openDatabase", () => {
test("creates and configures the migrated database", async () => {
const home = await createHome();
const database = openDatabase(home);
try {
expect(existsSync(join(home, "paseo.db"))).toBe(true);
expect(database.prepare("PRAGMA journal_mode").get()).toMatchObject({ journal_mode: "wal" });
expect(database.prepare("PRAGMA foreign_keys").get()).toMatchObject({ foreign_keys: 1 });
expect(database.prepare("PRAGMA busy_timeout").get()).toMatchObject({ timeout: 5_000 });
expect(database.prepare("PRAGMA user_version").get()).toMatchObject({ user_version: 3 });
expect(
database.prepare("SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name").all(),
).toEqual(
expect.arrayContaining([
expect.objectContaining({ name: "legacy_imports" }),
expect.objectContaining({ name: "push_tokens" }),
expect.objectContaining({ name: "schedules" }),
]),
);
} finally {
database.close();
}
});
test("rejects runtimes older than Node 22.20", () => {
expect(() => assertSupportedNodeVersion("22.19.9")).toThrow("Paseo requires Node.js >=22.20.0");
expect(() => assertSupportedNodeVersion("22.20.0")).not.toThrow();
expect(() => assertSupportedNodeVersion("23.0.0")).not.toThrow();
});
test("rolls back failed transactions", async () => {
const home = await createHome();
const database = openDatabase(home);
try {
expect(() => insertThenFail(database)).toThrow("stop");
expect(database.prepare("SELECT count(*) AS count FROM push_tokens").get()).toMatchObject({
count: 0,
});
} finally {
database.close();
}
});
});
+70
View File
@@ -0,0 +1,70 @@
import { chmodSync, mkdirSync } from "node:fs";
import { join } from "node:path";
// node:sqlite emits an ExperimentalWarning on Node 22. We deliberately accept it instead of
// changing process-wide warning behavior; the startup smoke checks below pin the features we use.
import { DatabaseSync } from "node:sqlite";
import { PRIVATE_FILE_MODE } from "../private-files.js";
import { runMigrations } from "./migrations.js";
const MINIMUM_NODE_VERSION = [22, 20, 0] as const;
const BUSY_TIMEOUT_MS = 5_000;
function parseNodeVersion(version: string): readonly number[] {
const match = /^(\d+)\.(\d+)\.(\d+)/.exec(version);
if (!match) {
throw new Error(`Unable to determine Node.js version from ${JSON.stringify(version)}`);
}
return [Number(match[1]), Number(match[2]), Number(match[3])];
}
export function assertSupportedNodeVersion(version = process.versions.node): void {
const actual = parseNodeVersion(version);
for (let index = 0; index < MINIMUM_NODE_VERSION.length; index += 1) {
if (actual[index] > MINIMUM_NODE_VERSION[index]) {
return;
}
if (actual[index] < MINIMUM_NODE_VERSION[index]) {
throw new Error(`Paseo requires Node.js >=22.20.0 for SQLite support; running ${version}`);
}
}
}
function readJournalMode(database: DatabaseSync): string {
const row = database.prepare("PRAGMA journal_mode").get() as { journal_mode: string };
return row.journal_mode.toLowerCase();
}
function verifyFts5(database: DatabaseSync): void {
try {
database.exec("CREATE VIRTUAL TABLE temp.paseo_fts5_smoke USING fts5(content)");
database.exec("DROP TABLE temp.paseo_fts5_smoke");
} catch (error) {
throw new Error("This Node.js runtime does not provide the required SQLite FTS5 support", {
cause: error,
});
}
}
export function openDatabase(paseoHome: string): DatabaseSync {
assertSupportedNodeVersion();
mkdirSync(paseoHome, { recursive: true });
const databasePath = join(paseoHome, "paseo.db");
const database = new DatabaseSync(databasePath);
try {
chmodSync(databasePath, PRIVATE_FILE_MODE);
database.exec(`
PRAGMA journal_mode = WAL;
PRAGMA foreign_keys = ON;
PRAGMA busy_timeout = ${BUSY_TIMEOUT_MS};
`);
if (readJournalMode(database) !== "wal") {
throw new Error("paseo.db could not enable required WAL journal mode");
}
verifyFts5(database);
runMigrations(database);
return database;
} catch (error) {
database.close();
throw new Error(`Failed to initialize SQLite database at ${databasePath}`, { cause: error });
}
}
@@ -0,0 +1,13 @@
import type { DatabaseSync } from "node:sqlite";
export function runInTransaction<T>(database: DatabaseSync, operation: () => T): T {
database.exec("BEGIN IMMEDIATE");
try {
const result = operation();
database.exec("COMMIT");
return result;
} catch (error) {
database.exec("ROLLBACK");
throw error;
}
}
+27 -7
View File
@@ -1,10 +1,13 @@
import { mkdtempSync, readFileSync, rmSync } from "node:fs";
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import type { DatabaseSync } from "node:sqlite";
import type pino from "pino";
import { afterEach, describe, expect, test } from "vitest";
import { openDatabase } from "../db/open.js";
import { createPushNotifications } from "./index.js";
import { PushTokenStore } from "./token-store.js";
function createLogger(): pino.Logger {
const logger = {
@@ -19,8 +22,10 @@ function createLogger(): pino.Logger {
describe("push notifications", () => {
const homes: string[] = [];
const databases: DatabaseSync[] = [];
afterEach(() => {
for (const database of databases.splice(0)) database.close();
for (const home of homes.splice(0)) {
rmSync(home, { recursive: true, force: true });
}
@@ -29,13 +34,19 @@ describe("push notifications", () => {
test("an offline device stops receiving notifications after 48 hours", async () => {
const home = mkdtempSync(path.join(tmpdir(), "paseo-push-notifications-"));
homes.push(home);
const filePath = path.join(home, "push-tokens.json");
const database = openDatabase(home);
databases.push(database);
let now = Date.parse("2026-08-10T00:00:00.000Z");
const deliveries: string[][] = [];
const store = new PushTokenStore({
database,
legacyFilePath: path.join(home, "push-tokens.json"),
logger: createLogger(),
now: () => now,
});
const pushNotifications = createPushNotifications({
logger: createLogger(),
filePath,
now: () => now,
store,
deliver: async (tokens) => deliveries.push(tokens),
});
@@ -44,17 +55,26 @@ describe("push notifications", () => {
await pushNotifications.send({ title: "Agent finished", body: "Done" });
expect(deliveries).toEqual([]);
expect(JSON.parse(readFileSync(filePath, "utf8"))).toEqual({ subscriptions: [] });
expect(database.prepare("SELECT count(*) AS count FROM push_tokens").get()).toMatchObject({
count: 0,
});
});
test("online revocation stops notifications immediately", async () => {
const home = mkdtempSync(path.join(tmpdir(), "paseo-push-notifications-"));
homes.push(home);
const database = openDatabase(home);
databases.push(database);
const deliveries: string[][] = [];
const store = new PushTokenStore({
database,
legacyFilePath: path.join(home, "push-tokens.json"),
logger: createLogger(),
now: () => Date.parse("2026-08-10T00:00:00.000Z"),
});
const pushNotifications = createPushNotifications({
logger: createLogger(),
filePath: path.join(home, "push-tokens.json"),
now: () => Date.parse("2026-08-10T00:00:00.000Z"),
store,
deliver: async (tokens) => deliveries.push(tokens),
});
+6 -11
View File
@@ -1,12 +1,10 @@
import type pino from "pino";
import { PushService, type PushPayload } from "./push-service.js";
import { PushTokenStore } from "./token-store.js";
import type { PushTokenStore } from "./token-store.js";
export type { PushPayload };
const PUSH_TOKEN_LEASE_MS = 48 * 60 * 60 * 1000;
export interface PushNotifications {
renew(token: string): void;
revoke(token: string): void;
@@ -17,26 +15,23 @@ export type PushNotificationSender = Pick<PushNotifications, "send">;
export function createPushNotifications(options: {
logger: pino.Logger;
filePath: string;
now?: () => number;
store: PushTokenStore;
deliver?: (tokens: string[], payload: PushPayload) => Promise<void>;
}): PushNotifications {
const now = options.now ?? Date.now;
const store = new PushTokenStore(options.logger, options.filePath, now, PUSH_TOKEN_LEASE_MS);
const service = new PushService(options.logger, (token) => store.revokeToken(token));
const service = new PushService(options.logger, (token) => options.store.revokeToken(token));
const deliver =
options.deliver ??
((tokens: string[], payload: PushPayload) => service.sendPush(tokens, payload));
return {
renew(token) {
store.renewToken(token);
options.store.renewToken(token);
},
revoke(token) {
store.revokeToken(token);
options.store.revokeToken(token);
},
async send(payload) {
const tokens = store.getActiveTokens();
const tokens = options.store.getActiveTokens();
options.logger.info({ tokenCount: tokens.length }, "Sending push notification");
if (tokens.length === 0) return;
await deliver(tokens, payload);
@@ -1,22 +1,22 @@
import { chmodSync, mkdtempSync, rmSync, statSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import type { DatabaseSync } from "node:sqlite";
import type pino from "pino";
import { describe, expect, test } from "vitest";
import { describe, expect, test, vi } from "vitest";
import { openDatabase } from "../db/open.js";
import { PRIVATE_FILE_MODE } from "../private-files.js";
import { createPushNotifications } from "./index.js";
import { PushTokenStore } from "./token-store.js";
const MODE_MASK = 0o777;
const PERMISSIVE_FILE_MODE = 0o644;
function createLogger(): pino.Logger {
function createLogger(warn: pino.Logger["warn"] = () => undefined): pino.Logger {
const logger = {
child: () => logger,
debug: () => undefined,
info: () => undefined,
warn: () => undefined,
warn,
error: () => undefined,
};
return logger as unknown as pino.Logger;
@@ -26,42 +26,128 @@ function modeOf(filePath: string): number {
return statSync(filePath).mode & MODE_MASK;
}
describe.skipIf(process.platform === "win32")("PushTokenStore file permissions", () => {
test("persists push tokens with private permissions", () => {
const home = mkdtempSync(path.join(tmpdir(), "paseo-push-tokens-"));
const tokenPath = path.join(home, "push-tokens.json");
try {
const pushNotifications = createPushNotifications({
logger: createLogger(),
filePath: tokenPath,
});
function createStore(
database: DatabaseSync,
tokenPath: string,
logger = createLogger(),
): PushTokenStore {
return new PushTokenStore({ database, legacyFilePath: tokenPath, logger });
}
pushNotifications.renew("ExponentPushToken[test]");
describe.skipIf(process.platform === "win32")("PushTokenStore file permissions", () => {
test("persists push tokens in a private database", () => {
const home = mkdtempSync(path.join(tmpdir(), "paseo-push-tokens-"));
const tokenPath = path.join(home, "paseo.db");
const database = openDatabase(home);
try {
const store = createStore(database, path.join(home, "push-tokens.json"));
store.renewToken("ExponentPushToken[test]");
expect(modeOf(tokenPath)).toBe(PRIVATE_FILE_MODE);
} finally {
database.close();
rmSync(home, { recursive: true, force: true });
}
});
test("repairs existing push token file permissions when loading", async () => {
test("repairs existing push token file permissions when importing", () => {
const home = mkdtempSync(path.join(tmpdir(), "paseo-push-tokens-"));
const tokenPath = path.join(home, "push-tokens.json");
writeFileSync(tokenPath, JSON.stringify({ tokens: ["ExponentPushToken[test]"] }));
chmodSync(tokenPath, PERMISSIVE_FILE_MODE);
const database = openDatabase(home);
try {
writeFileSync(tokenPath, JSON.stringify({ tokens: ["ExponentPushToken[test]"] }));
chmodSync(tokenPath, PERMISSIVE_FILE_MODE);
const store = createStore(database, tokenPath);
const deliveries: string[][] = [];
const pushNotifications = createPushNotifications({
logger: createLogger(),
filePath: tokenPath,
deliver: async (tokens) => deliveries.push(tokens),
});
await pushNotifications.send({ title: "Agent finished", body: "Done" });
expect(deliveries).toEqual([["ExponentPushToken[test]"]]);
expect(store.getActiveTokens()).toEqual(["ExponentPushToken[test]"]);
expect(modeOf(tokenPath)).toBe(PRIVATE_FILE_MODE);
} finally {
database.close();
rmSync(home, { recursive: true, force: true });
}
});
});
describe("PushTokenStore legacy import", () => {
test("imports once and records the marker", () => {
const home = mkdtempSync(path.join(tmpdir(), "paseo-push-tokens-"));
const tokenPath = path.join(home, "push-tokens.json");
writeFileSync(
tokenPath,
JSON.stringify({
subscriptions: [
{
token: "ExponentPushToken[legacy]",
expiresAt: "2099-08-12T00:00:00.000Z",
},
],
}),
);
const database = openDatabase(home);
try {
const store = createStore(database, tokenPath);
expect(store.getActiveTokens()).toEqual(["ExponentPushToken[legacy]"]);
store.revokeToken("ExponentPushToken[legacy]");
const reloaded = createStore(database, tokenPath);
expect(reloaded.getActiveTokens()).toEqual([]);
expect(
database
.prepare("SELECT imported_count FROM legacy_imports WHERE store = ?")
.get("push_tokens"),
).toMatchObject({ imported_count: 1 });
} finally {
database.close();
rmSync(home, { recursive: true, force: true });
}
});
test("skips and logs invalid legacy tokens", () => {
const home = mkdtempSync(path.join(tmpdir(), "paseo-push-tokens-"));
const tokenPath = path.join(home, "push-tokens.json");
writeFileSync(tokenPath, JSON.stringify({ tokens: ["ExponentPushToken[valid]", " ", 42] }));
const database = openDatabase(home);
const warn = vi.fn();
try {
const store = createStore(database, tokenPath, createLogger(warn));
expect(store.getActiveTokens()).toEqual(["ExponentPushToken[valid]"]);
expect(warn).toHaveBeenCalledTimes(2);
expect(
database
.prepare("SELECT skipped_count FROM legacy_imports WHERE store = ?")
.get("push_tokens"),
).toMatchObject({ skipped_count: 2 });
} finally {
database.close();
rmSync(home, { recursive: true, force: true });
}
});
test("rolls back the whole legacy import when its marker cannot be written", () => {
const home = mkdtempSync(path.join(tmpdir(), "paseo-push-tokens-"));
const tokenPath = path.join(home, "push-tokens.json");
writeFileSync(tokenPath, JSON.stringify({ tokens: ["ExponentPushToken[legacy]"] }));
const database = openDatabase(home);
database.exec(`
CREATE TRIGGER reject_legacy_import_marker
BEFORE INSERT ON legacy_imports
BEGIN
SELECT RAISE(ABORT, 'marker rejected');
END;
`);
try {
expect(() => createStore(database, tokenPath)).toThrow("Failed to import legacy push tokens");
expect(database.prepare("SELECT count(*) AS count FROM push_tokens").get()).toMatchObject({
count: 0,
});
expect(database.prepare("SELECT count(*) AS count FROM legacy_imports").get()).toMatchObject({
count: 0,
});
} finally {
database.close();
rmSync(home, { recursive: true, force: true });
}
});
@@ -69,36 +155,28 @@ describe.skipIf(process.platform === "win32")("PushTokenStore file permissions",
describe("PushTokenStore persistence", () => {
test("does not apply a revocation until the updated subscriptions are durable", () => {
let rejectWrites = false;
let persisted = "";
const store = new PushTokenStore(
createLogger(),
path.join(tmpdir(), `missing-push-token-store-${process.pid}`, "push-tokens.json"),
() => Date.parse("2026-08-10T00:00:00.000Z"),
48 * 60 * 60 * 1000,
(_filePath, payload) => {
if (rejectWrites) throw new Error("disk full");
persisted = String(payload);
},
);
const home = mkdtempSync(path.join(tmpdir(), "paseo-push-tokens-"));
const database = openDatabase(home);
try {
const store = createStore(database, path.join(home, "push-tokens.json"));
store.renewToken("ExponentPushToken[test]");
database.exec(`
CREATE TRIGGER reject_push_token_delete
BEFORE DELETE ON push_tokens
BEGIN
SELECT RAISE(ABORT, 'delete rejected');
END;
`);
store.renewToken("ExponentPushToken[test]");
rejectWrites = true;
expect(() => store.revokeToken("ExponentPushToken[test]")).toThrow("delete rejected");
expect(store.getActiveTokens()).toEqual(["ExponentPushToken[test]"]);
expect(() => store.revokeToken("ExponentPushToken[test]")).toThrow("disk full");
expect(store.getActiveTokens()).toEqual(["ExponentPushToken[test]"]);
expect(JSON.parse(persisted)).toEqual({
subscriptions: [
{
token: "ExponentPushToken[test]",
expiresAt: "2026-08-12T00:00:00.000Z",
},
],
});
rejectWrites = false;
store.revokeToken("ExponentPushToken[test]");
expect(store.getActiveTokens()).toEqual([]);
expect(JSON.parse(persisted)).toEqual({ subscriptions: [] });
database.exec("DROP TRIGGER reject_push_token_delete");
store.revokeToken("ExponentPushToken[test]");
expect(store.getActiveTokens()).toEqual([]);
} finally {
database.close();
rmSync(home, { recursive: true, force: true });
}
});
});
+140 -103
View File
@@ -1,141 +1,178 @@
import type pino from "pino";
import { existsSync, readFileSync } from "node:fs";
import type { DatabaseSync } from "node:sqlite";
import type pino from "pino";
import { z } from "zod";
import { ensurePrivateFile, writePrivateFileAtomicSync } from "../private-files.js";
import { hasLegacyImportMarker, recordLegacyImportMarker } from "../db/legacy-imports.js";
import { runInTransaction } from "../db/transaction.js";
import { ensurePrivateFile } from "../private-files.js";
/**
* Store for Expo push tokens.
*
* Tokens are persisted to disk so pushes still work after daemon restarts.
*/
const LEGACY_IMPORT_STORE = "push_tokens";
const DEFAULT_LEASE_MS = 48 * 60 * 60 * 1000;
const PushTokenRecordSchema = z.object({
token: z.string().trim().min(1),
expiresAt: z.string().refine((value) => Number.isFinite(Date.parse(value))),
});
interface PushTokenStoreOptions {
database: DatabaseSync;
legacyFilePath: string;
logger: pino.Logger;
now?: () => number;
leaseMs?: number;
}
interface PushTokenRow {
payload: string;
}
/** Store for leased Expo push-token subscriptions. */
export class PushTokenStore {
private readonly database: DatabaseSync;
private readonly logger: pino.Logger;
private subscriptions = new Map<string, number>();
private readonly filePath: string;
private readonly legacyFilePath: string;
private readonly now: () => number;
private readonly leaseMs: number;
private readonly write: typeof writePrivateFileAtomicSync;
constructor(
logger: pino.Logger,
filePath: string,
now: () => number,
leaseMs: number,
write: typeof writePrivateFileAtomicSync = writePrivateFileAtomicSync,
) {
this.logger = logger.child({ component: "token-store" });
this.filePath = filePath;
this.now = now;
this.leaseMs = leaseMs;
this.write = write;
this.loadFromDisk();
constructor(options: PushTokenStoreOptions) {
this.database = options.database;
this.logger = options.logger.child({ component: "token-store" });
this.legacyFilePath = options.legacyFilePath;
this.now = options.now ?? Date.now;
this.leaseMs = options.leaseMs ?? DEFAULT_LEASE_MS;
this.importLegacyTokens();
}
renewToken(token: string): void {
const normalized = token.trim();
if (!normalized) return;
const current = this.readToken(normalized);
const now = this.now();
const currentExpiry = this.subscriptions.get(normalized);
if (currentExpiry !== undefined && currentExpiry - now > this.leaseMs / 2) return;
const next = new Map(this.subscriptions);
next.set(normalized, now + this.leaseMs);
this.persist(next);
this.subscriptions = next;
this.logger.debug({ total: this.subscriptions.size }, "Renewed token");
if (current && Date.parse(current.expiresAt) - now > this.leaseMs / 2) return;
const record = {
token: normalized,
expiresAt: new Date(now + this.leaseMs).toISOString(),
};
this.database
.prepare(
"INSERT INTO push_tokens (token, payload) VALUES (?, ?) " +
"ON CONFLICT(token) DO UPDATE SET payload = excluded.payload",
)
.run(record.token, JSON.stringify(record));
this.logger.debug({ total: this.count() }, "Renewed token");
}
revokeToken(token: string): void {
const normalized = token.trim();
if (!normalized) return;
if (!this.subscriptions.has(normalized)) return;
const next = new Map(this.subscriptions);
next.delete(normalized);
this.persist(next);
this.subscriptions = next;
this.logger.debug({ total: this.subscriptions.size }, "Revoked token");
const result = this.database.prepare("DELETE FROM push_tokens WHERE token = ?").run(normalized);
if (result.changes > 0) {
this.logger.debug({ total: this.count() }, "Revoked token");
}
}
getActiveTokens(): string[] {
const now = this.now();
const active = new Map(this.subscriptions);
for (const [token, expiresAt] of this.subscriptions) {
if (expiresAt <= now) {
active.delete(token);
const rows = this.database
.prepare("SELECT payload FROM push_tokens ORDER BY rowid")
.all() as unknown as PushTokenRow[];
const active: string[] = [];
const expired: string[] = [];
for (const row of rows) {
const record = PushTokenRecordSchema.parse(JSON.parse(row.payload));
if (Date.parse(record.expiresAt) <= now) {
expired.push(record.token);
} else {
active.push(record.token);
}
}
if (active.size !== this.subscriptions.size) {
if (expired.length > 0) {
try {
this.persist(active);
this.subscriptions = active;
runInTransaction(this.database, () => {
const remove = this.database.prepare("DELETE FROM push_tokens WHERE token = ?");
for (const token of expired) remove.run(token);
});
} catch {
// Keep the previous state so a later send retries pruning. Expired tokens
// are still excluded from this delivery.
// Expired tokens stay excluded and a later delivery retries pruning them.
}
}
return Array.from(active.keys());
return active;
}
private loadFromDisk(): void {
try {
if (!existsSync(this.filePath)) {
return;
}
ensurePrivateFile(this.filePath);
const raw = readFileSync(this.filePath, "utf-8");
const parsed = JSON.parse(raw) as { subscriptions?: unknown; tokens?: unknown };
const loaded = new Map<string, number>();
const subscriptions = Array.isArray(parsed.subscriptions) ? parsed.subscriptions : [];
for (const value of subscriptions) {
if (!value || typeof value !== "object") continue;
const candidate = value as { token?: unknown; expiresAt?: unknown };
if (typeof candidate.token !== "string" || typeof candidate.expiresAt !== "string")
continue;
const token = candidate.token.trim();
const expiresAt = Date.parse(candidate.expiresAt);
if (token && Number.isFinite(expiresAt)) {
loaded.set(token, expiresAt);
}
}
this.subscriptions = loaded;
const legacyTokens = Array.isArray(parsed.tokens)
? parsed.tokens.filter((token): token is string => typeof token === "string")
: [];
if (legacyTokens.length > 0) {
const migrated = new Map(loaded);
const expiresAt = this.now() + this.leaseMs;
for (const token of legacyTokens) {
const normalized = token.trim();
if (normalized) migrated.set(normalized, expiresAt);
}
this.persist(migrated);
this.subscriptions = migrated;
}
this.logger.info({ total: this.subscriptions.size }, "Loaded push tokens");
} catch (error) {
const err = error instanceof Error ? error : new Error(String(error));
this.logger.warn({ err }, "Failed to load push tokens");
}
private readToken(token: string): z.infer<typeof PushTokenRecordSchema> | null {
const row = this.database
.prepare("SELECT payload FROM push_tokens WHERE token = ?")
.get(token) as PushTokenRow | undefined;
return row ? PushTokenRecordSchema.parse(JSON.parse(row.payload)) : null;
}
private persist(subscriptions: ReadonlyMap<string, number>): void {
private count(): number {
const row = this.database.prepare("SELECT count(*) AS count FROM push_tokens").get() as {
count: number;
};
return row.count;
}
private importLegacyTokens(): void {
if (hasLegacyImportMarker(this.database, LEGACY_IMPORT_STORE)) return;
try {
const payload =
JSON.stringify(
{
subscriptions: Array.from(subscriptions, ([token, expiresAt]) => ({
token,
expiresAt: new Date(expiresAt).toISOString(),
})),
},
null,
2,
) + "\n";
this.write(this.filePath, payload);
runInTransaction(this.database, () => {
let importedCount = 0;
let skippedCount = 0;
if (this.count() === 0 && existsSync(this.legacyFilePath)) {
ensurePrivateFile(this.legacyFilePath);
const raw = readFileSync(this.legacyFilePath, "utf-8");
let parsed: unknown;
try {
parsed = JSON.parse(raw);
} catch (error) {
this.logger.warn(
{ err: error, filePath: this.legacyFilePath },
"Skipping invalid legacy push token file",
);
skippedCount += 1;
}
const source = parsed && typeof parsed === "object" ? parsed : {};
const subscriptions = Array.isArray((source as { subscriptions?: unknown }).subscriptions)
? (source as { subscriptions: unknown[] }).subscriptions
: [];
const tokens = Array.isArray((source as { tokens?: unknown }).tokens)
? (source as { tokens: unknown[] }).tokens.map((token) => ({
token,
expiresAt: new Date(this.now() + this.leaseMs).toISOString(),
}))
: [];
for (const candidate of [...subscriptions, ...tokens]) {
const record = PushTokenRecordSchema.safeParse(candidate);
if (!record.success) {
skippedCount += 1;
this.logger.warn(
{ filePath: this.legacyFilePath, err: record.error },
"Skipping invalid legacy push token",
);
continue;
}
const result = this.database
.prepare("INSERT OR IGNORE INTO push_tokens (token, payload) VALUES (?, ?)")
.run(record.data.token, JSON.stringify(record.data));
importedCount += Number(result.changes);
}
}
recordLegacyImportMarker(this.database, LEGACY_IMPORT_STORE, {
importedCount,
skippedCount,
});
this.logger.info({ importedCount, skippedCount }, "Legacy push token import complete");
});
} catch (error) {
const err = error instanceof Error ? error : new Error(String(error));
this.logger.warn({ err }, "Failed to persist push tokens");
throw err;
throw new Error(`Failed to import legacy push tokens from ${this.legacyFilePath}`, {
cause: error,
});
}
}
}
@@ -1,6 +1,7 @@
import { mkdir, mkdtemp, rm } from "node:fs/promises";
import { join } from "node:path";
import { tmpdir } from "node:os";
import type { DatabaseSync } from "node:sqlite";
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
import { AgentManager } from "../agent/agent-manager.js";
import { AgentStorage } from "../agent/agent-storage.js";
@@ -41,6 +42,7 @@ import {
} from "./service.js";
import { ScheduleStore } from "./store.js";
import type { ScheduleExecutionResult, StoredSchedule } from "@getpaseo/protocol/schedule/types";
import { openDatabase } from "../db/open.js";
interface ScheduleServiceInternals {
executeSchedule(schedule: StoredSchedule, runId: string): Promise<ScheduleExecutionResult>;
@@ -75,10 +77,15 @@ const TEST_CLAUDE_PROVIDER_DEFINITION = {
};
let workspaceArchiveInProgress = false;
let database: DatabaseSync;
type TestScheduleServiceOptions = Omit<
ScheduleServiceOptions,
"createAgent" | "createDirectoryWorkspace" | "createPaseoWorktreeWorkspace" | "archiveWorkspace"
| "database"
| "createAgent"
| "createDirectoryWorkspace"
| "createPaseoWorktreeWorkspace"
| "archiveWorkspace"
> & {
agentManager: AgentManager;
providerSnapshotManager: Pick<ProviderSnapshotManager, "resolveCreateConfig">;
@@ -159,6 +166,7 @@ function createScheduleService(options: TestScheduleServiceOptions): ScheduleSer
};
return new ScheduleService({
...options,
database,
createAgent:
options.createAgent ??
((input) =>
@@ -300,6 +308,7 @@ describe("ScheduleService", () => {
beforeEach(async () => {
tempDir = await mkdtemp(join(tmpdir(), "schedule-service-test-"));
database = openDatabase(tempDir);
await mkdir(join(tempDir, "agents"), { recursive: true });
agentStorage = new AgentStorage(join(tempDir, "agents"), createTestLogger());
await agentStorage.initialize();
@@ -310,6 +319,7 @@ describe("ScheduleService", () => {
// Drain pending background persists before deleting the dir to avoid
// ENOTEMPTY races when AgentManager flushes a snapshot mid-cleanup.
await agentStorage.flush();
database.close();
await rm(tempDir, { recursive: true, force: true });
});
@@ -768,7 +778,11 @@ describe("ScheduleService", () => {
const runStarted = new Promise<void>((resolve) => {
releaseRun = resolve;
});
const store = new ScheduleStore(join(tempDir, "schedules"));
const store = new ScheduleStore({
database,
legacyDir: join(tempDir, "schedules"),
logger: createTestLogger(),
});
const legacy = await store.create({
name: null,
prompt: "finish/update race",
@@ -1891,7 +1905,11 @@ describe("ScheduleService", () => {
const interruptedAt = now.toISOString();
const associatedAgentId = "11111111-1111-4111-8111-111111111111";
const workspaceId = "wks_interrupted_with_agent";
const store = new ScheduleStore(join(tempDir, "schedules"));
const store = new ScheduleStore({
database,
legacyDir: join(tempDir, "schedules"),
logger: createTestLogger(),
});
await store.update(created.id, (schedule) => ({
...schedule,
runs: [
@@ -1959,7 +1977,11 @@ describe("ScheduleService", () => {
const interruptedAt = now.toISOString();
const workspaceId = "wks_interrupted_without_agent";
const store = new ScheduleStore(join(tempDir, "schedules"));
const store = new ScheduleStore({
database,
legacyDir: join(tempDir, "schedules"),
logger: createTestLogger(),
});
await store.update(created.id, (schedule) => ({
...schedule,
runs: [
@@ -1,6 +1,7 @@
import { randomUUID } from "node:crypto";
import { stat } from "node:fs/promises";
import { join } from "node:path";
import type { DatabaseSync } from "node:sqlite";
import type { Logger } from "pino";
import type { AgentManager } from "../agent/agent-manager.js";
import type { AgentSessionConfig } from "../agent/agent-sdk-types.js";
@@ -215,6 +216,7 @@ interface ScheduleWorkspaceCreateInput {
export interface ScheduleServiceOptions {
paseoHome: string;
database: DatabaseSync;
logger: Logger;
agentManager: ScheduleAgentManager;
agentStorage: AgentStorage;
@@ -252,7 +254,11 @@ export class ScheduleService {
private tickTimer: ReturnType<typeof setInterval> | null = null;
constructor(options: ScheduleServiceOptions) {
this.store = new ScheduleStore(join(options.paseoHome, "schedules"));
this.store = new ScheduleStore({
database: options.database,
legacyDir: join(options.paseoHome, "schedules"),
logger: options.logger,
});
this.logger = options.logger.child({ module: "schedule-service" });
this.agentManager = options.agentManager;
this.agentStorage = options.agentStorage;
@@ -1,19 +1,43 @@
import { mkdtemp, rm } from "node:fs/promises";
import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
import type { DatabaseSync } from "node:sqlite";
import type pino from "pino";
import { join } from "node:path";
import { tmpdir } from "node:os";
import { afterEach, beforeEach, describe, expect, test } from "vitest";
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
import { openDatabase } from "../db/open.js";
import { createTestLogger } from "../../test-utils/test-logger.js";
import { ScheduleStore } from "./store.js";
describe("ScheduleStore", () => {
let tempDir: string;
let database: DatabaseSync;
let store: ScheduleStore;
function createStore(logger = createTestLogger()): ScheduleStore {
return new ScheduleStore({
database,
legacyDir: join(tempDir, "schedules"),
logger,
});
}
function createRecordingLogger(warn: pino.Logger["warn"]): pino.Logger {
const logger = {
child: () => logger,
info: () => undefined,
warn,
};
return logger as unknown as pino.Logger;
}
beforeEach(async () => {
tempDir = await mkdtemp(join(tmpdir(), "schedule-store-test-"));
store = new ScheduleStore(tempDir);
database = openDatabase(tempDir);
store = createStore();
});
afterEach(async () => {
database.close();
await rm(tempDir, { recursive: true, force: true });
});
@@ -40,7 +64,7 @@ describe("ScheduleStore", () => {
runs: [],
});
const reloaded = new ScheduleStore(tempDir);
const reloaded = createStore();
const listed = await reloaded.list();
expect(created.id).toHaveLength(8);
@@ -81,7 +105,7 @@ describe("ScheduleStore", () => {
};
await store.update(created.id, () => updated);
const reloaded = await new ScheduleStore(tempDir).get(created.id);
const reloaded = await createStore().get(created.id);
expect(reloaded).toEqual(updated);
});
@@ -182,7 +206,7 @@ describe("ScheduleStore", () => {
prompt: "after",
runs: [{ id: "run-1" }],
});
await expect(new ScheduleStore(tempDir).get(created.id)).resolves.toMatchObject({
await expect(createStore().get(created.id)).resolves.toMatchObject({
prompt: "after",
runs: [{ id: "run-1" }],
});
@@ -211,7 +235,11 @@ describe("ScheduleStore", () => {
}
}
const gatedStore = new GatedListScheduleStore(tempDir);
const gatedStore = new GatedListScheduleStore({
database,
legacyDir: join(tempDir, "schedules"),
logger: createTestLogger(),
});
const target = {
type: "new-agent" as const,
config: { provider: "claude" as const, cwd: tempDir },
@@ -304,4 +332,99 @@ describe("ScheduleStore", () => {
});
expect(await gatedStore.list()).toHaveLength(2);
});
test("imports legacy schedules once and records the marker", async () => {
const legacyDir = join(tempDir, "schedules");
await mkdir(legacyDir, { recursive: true });
const legacy = {
id: "legacy01",
name: "Legacy",
prompt: "Imported",
cadence: { type: "every" as const, everyMs: 60_000 },
target: {
type: "new-agent" as const,
config: { provider: "claude" as const, cwd: tempDir },
},
status: "active" as const,
createdAt: "2026-01-01T00:00:00.000Z",
updatedAt: "2026-01-01T00:00:00.000Z",
nextRunAt: "2026-01-01T00:01:00.000Z",
lastRunAt: null,
pausedAt: null,
expiresAt: null,
maxRuns: null,
runs: [],
};
await writeFile(join(legacyDir, "legacy01.json"), JSON.stringify(legacy));
database.prepare("DELETE FROM legacy_imports WHERE store = ?").run("schedules");
const imported = createStore();
await expect(imported.get(legacy.id)).resolves.toEqual(legacy);
await imported.delete(legacy.id);
createStore();
await expect(imported.get(legacy.id)).resolves.toBeNull();
expect(
database
.prepare("SELECT imported_count FROM legacy_imports WHERE store = ?")
.get("schedules"),
).toMatchObject({ imported_count: 1 });
});
test("skips and logs invalid legacy schedule records", async () => {
const legacyDir = join(tempDir, "schedules");
await mkdir(legacyDir, { recursive: true });
await writeFile(join(legacyDir, "invalid.json"), "{not-json");
database.prepare("DELETE FROM legacy_imports WHERE store = ?").run("schedules");
const warn = vi.fn();
createStore(createRecordingLogger(warn));
expect(warn).toHaveBeenCalledWith(
expect.objectContaining({ filePath: join(legacyDir, "invalid.json") }),
"Skipping invalid legacy schedule record",
);
expect(
database.prepare("SELECT skipped_count FROM legacy_imports WHERE store = ?").get("schedules"),
).toMatchObject({ skipped_count: 1 });
});
test("rolls back the whole legacy schedule import on failure", async () => {
const legacyDir = join(tempDir, "schedules");
await mkdir(legacyDir, { recursive: true });
const legacy = {
id: "duplicate",
name: null,
prompt: "Imported",
cadence: { type: "every" as const, everyMs: 60_000 },
target: {
type: "new-agent" as const,
config: { provider: "claude" as const, cwd: tempDir },
},
status: "active" as const,
createdAt: "2026-01-01T00:00:00.000Z",
updatedAt: "2026-01-01T00:00:00.000Z",
nextRunAt: "2026-01-01T00:01:00.000Z",
lastRunAt: null,
pausedAt: null,
expiresAt: null,
maxRuns: null,
runs: [],
};
await Promise.all([
writeFile(join(legacyDir, "one.json"), JSON.stringify(legacy)),
writeFile(join(legacyDir, "two.json"), JSON.stringify(legacy)),
]);
database.prepare("DELETE FROM legacy_imports WHERE store = ?").run("schedules");
expect(() => createStore()).toThrow("Failed to import legacy schedules");
expect(database.prepare("SELECT count(*) AS count FROM schedules").get()).toMatchObject({
count: 0,
});
expect(
database
.prepare("SELECT count(*) AS count FROM legacy_imports WHERE store = ?")
.get("schedules"),
).toMatchObject({ count: 0 });
});
});
+118 -36
View File
@@ -1,12 +1,17 @@
import { randomBytes } from "node:crypto";
import { mkdir, readFile, readdir, rm } from "node:fs/promises";
import { existsSync, readFileSync, readdirSync } from "node:fs";
import { join } from "node:path";
import type { DatabaseSync } from "node:sqlite";
import {
StoredScheduleSchema,
type ScheduleTarget,
type StoredSchedule,
} from "@getpaseo/protocol/schedule/types";
import { writeJsonFileAtomic } from "../atomic-file.js";
import type { Logger } from "pino";
import { hasLegacyImportMarker, recordLegacyImportMarker } from "../db/legacy-imports.js";
import { runInTransaction } from "../db/transaction.js";
const LEGACY_IMPORT_STORE = "schedules";
function generateScheduleId(): string {
return randomBytes(4).toString("hex");
@@ -19,6 +24,16 @@ interface ScheduleNameTargetUpsert {
update: ScheduleUpdater;
}
interface ScheduleStoreOptions {
database: DatabaseSync;
legacyDir: string;
logger: Logger;
}
interface ScheduleRow {
payload: string;
}
function canonicalize(value: unknown): unknown {
if (Array.isArray(value)) {
return value.map(canonicalize);
@@ -90,42 +105,29 @@ function matchesNameAndTarget(
export class ScheduleStore {
private readonly scheduleMutations = new Map<string, Promise<unknown>>();
private readonly identityMutations = new Map<string, Promise<unknown>>();
private readonly database: DatabaseSync;
private readonly legacyDir: string;
private readonly logger: Logger;
constructor(private readonly dir: string) {}
private filePath(id: string): string {
return join(this.dir, `${id}.json`);
}
private async ensureDir(): Promise<void> {
await mkdir(this.dir, { recursive: true });
constructor(options: ScheduleStoreOptions) {
this.database = options.database;
this.legacyDir = options.legacyDir;
this.logger = options.logger.child({ component: "schedule-store" });
this.importLegacySchedules();
}
async list(): Promise<StoredSchedule[]> {
await this.ensureDir();
const entries = await readdir(this.dir, { withFileTypes: true });
const schedules = await Promise.all(
entries
.filter((entry) => entry.isFile() && entry.name.endsWith(".json"))
.map(async (entry) => {
const content = await readFile(join(this.dir, entry.name), "utf-8");
return StoredScheduleSchema.parse(JSON.parse(content));
}),
);
return schedules.sort((left, right) => left.createdAt.localeCompare(right.createdAt));
const rows = this.database
.prepare("SELECT payload FROM schedules ORDER BY created_at, id")
.all() as unknown as ScheduleRow[];
return rows.map((row) => StoredScheduleSchema.parse(JSON.parse(row.payload)));
}
async get(id: string): Promise<StoredSchedule | null> {
await this.ensureDir();
try {
const content = await readFile(this.filePath(id), "utf-8");
return StoredScheduleSchema.parse(JSON.parse(content));
} catch (error) {
if ((error as NodeJS.ErrnoException).code === "ENOENT") {
return null;
}
throw error;
}
const row = this.database.prepare("SELECT payload FROM schedules WHERE id = ?").get(id) as
| ScheduleRow
| undefined;
return row ? StoredScheduleSchema.parse(JSON.parse(row.payload)) : null;
}
async create(schedule: Omit<StoredSchedule, "id">): Promise<StoredSchedule> {
@@ -184,15 +186,39 @@ export class ScheduleStore {
});
}
private async write(schedule: StoredSchedule): Promise<void> {
await this.ensureDir();
await writeJsonFileAtomic(this.filePath(schedule.id), schedule);
private write(schedule: StoredSchedule): void {
const identityKey =
schedule.status !== "completed" && normalizeOptionalScheduleName(schedule.name) !== null
? nameTargetIdentityKey(schedule.name!, schedule.target)
: null;
this.database
.prepare(
`INSERT INTO schedules (id, created_at, identity_key, payload)
VALUES (?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
created_at = excluded.created_at,
identity_key = excluded.identity_key,
payload = excluded.payload`,
)
.run(schedule.id, schedule.createdAt, identityKey, JSON.stringify(schedule));
}
private insert(schedule: StoredSchedule): void {
const identityKey =
schedule.status !== "completed" && normalizeOptionalScheduleName(schedule.name) !== null
? nameTargetIdentityKey(schedule.name!, schedule.target)
: null;
this.database
.prepare(
`INSERT INTO schedules (id, created_at, identity_key, payload)
VALUES (?, ?, ?, ?)`,
)
.run(schedule.id, schedule.createdAt, identityKey, JSON.stringify(schedule));
}
async delete(id: string): Promise<void> {
await this.serializeScheduleMutation(id, async () => {
await this.ensureDir();
await rm(this.filePath(id), { force: true });
this.database.prepare("DELETE FROM schedules WHERE id = ?").run(id);
});
}
@@ -250,4 +276,60 @@ export class ScheduleStore {
return updated;
});
}
private count(): number {
const row = this.database.prepare("SELECT count(*) AS count FROM schedules").get() as {
count: number;
};
return row.count;
}
private importLegacySchedules(): void {
if (hasLegacyImportMarker(this.database, LEGACY_IMPORT_STORE)) {
return;
}
try {
runInTransaction(this.database, () => {
let importedCount = 0;
let skippedCount = 0;
if (this.count() === 0 && existsSync(this.legacyDir)) {
const entries = readdirSync(this.legacyDir, { withFileTypes: true });
for (const entry of entries) {
if (!entry.isFile() || !entry.name.endsWith(".json")) {
continue;
}
const filePath = join(this.legacyDir, entry.name);
const content = readFileSync(filePath, "utf-8");
let json: unknown;
try {
json = JSON.parse(content);
} catch (error) {
skippedCount += 1;
this.logger.warn({ err: error, filePath }, "Skipping invalid legacy schedule record");
continue;
}
const parsed = StoredScheduleSchema.safeParse(json);
if (!parsed.success) {
skippedCount += 1;
this.logger.warn(
{ err: parsed.error, filePath },
"Skipping invalid legacy schedule record",
);
continue;
}
this.insert(parsed.data);
importedCount += 1;
}
}
recordLegacyImportMarker(this.database, LEGACY_IMPORT_STORE, {
importedCount,
skippedCount,
});
this.logger.info({ importedCount, skippedCount }, "Legacy schedule import complete");
});
} catch (error) {
throw new Error(`Failed to import legacy schedules from ${this.legacyDir}`, { cause: error });
}
}
}
@@ -17,6 +17,7 @@ import { BrowserToolsBroker } from "./browser-tools/broker.js";
import type { CheckoutDiffManager } from "./checkout-diff-manager.js";
import type { DaemonConfigStore } from "./daemon-config-store.js";
import type { DownloadTokenStore } from "./file-download/token-store.js";
import type { PushTokenStore } from "./push/token-store.js";
import type { ScheduleService } from "./schedule/service.js";
import { createStub } from "./test-utils/class-mocks.js";
import { DaemonClient } from "./test-utils/daemon-client.js";
@@ -292,6 +293,7 @@ function createVoiceAssistantWebSocketServer(params: {
createStub<AgentManager>(agentManager),
createStub<AgentStorage>({}),
createStub<DownloadTokenStore>({}),
createStub<PushTokenStore>({}),
"/tmp/paseo-browser-tools-websocket-test",
createStub<DaemonConfigStore>(daemonConfigStore),
null,
@@ -4,6 +4,7 @@ import type pino from "pino";
import type { AgentManager } from "./agent/agent-manager.js";
import type { AgentStorage } from "./agent/agent-storage.js";
import type { DownloadTokenStore } from "./file-download/token-store.js";
import type { PushTokenStore } from "./push/token-store.js";
import type { DaemonConfigStore } from "./daemon-config-store.js";
import type { ScheduleService } from "./schedule/service.js";
import type { CheckoutDiffManager } from "./checkout-diff-manager.js";
@@ -110,6 +111,7 @@ function createServer(agentManagerOverrides?: Record<string, unknown>) {
createStub<AgentManager>(agentManager),
createStub<AgentStorage>({}),
createStub<DownloadTokenStore>({}),
createStub<PushTokenStore>({}),
"/tmp/paseo-test",
createStub<DaemonConfigStore>(daemonConfigStore),
null,
@@ -4,6 +4,7 @@ import type pino from "pino";
import type { AgentManager } from "./agent/agent-manager.js";
import type { AgentStorage } from "./agent/agent-storage.js";
import type { DownloadTokenStore } from "./file-download/token-store.js";
import type { PushTokenStore } from "./push/token-store.js";
import type { DaemonConfigStore } from "./daemon-config-store.js";
import type { ScheduleService } from "./schedule/service.js";
import type { CheckoutDiffManager } from "./checkout-diff-manager.js";
@@ -244,6 +245,7 @@ function createServer(options?: {
}),
createStub<AgentStorage>({}),
createStub<DownloadTokenStore>({}),
createStub<PushTokenStore>({}),
"/tmp/paseo-test",
createStub<DaemonConfigStore>(daemonConfigStore),
null,
@@ -4,6 +4,7 @@ import type pino from "pino";
import type { AgentManager } from "./agent/agent-manager.js";
import type { AgentStorage } from "./agent/agent-storage.js";
import type { DownloadTokenStore } from "./file-download/token-store.js";
import type { PushTokenStore } from "./push/token-store.js";
import type { DaemonConfigStore } from "./daemon-config-store.js";
import type { ScheduleService } from "./schedule/service.js";
import type { CheckoutDiffManager } from "./checkout-diff-manager.js";
@@ -141,6 +142,7 @@ function createServer(terminalManager: TerminalManager, workspaceRegistry?: Work
createStub<AgentManager>(agentManager),
createStub<AgentStorage>({}),
createStub<DownloadTokenStore>({}),
createStub<PushTokenStore>({}),
"/tmp/paseo-test",
createStub<DaemonConfigStore>(daemonConfigStore),
null,
@@ -1,6 +1,5 @@
import { WebSocket, WebSocketServer } from "ws";
import type { IncomingMessage, Server as HTTPServer } from "http";
import { join } from "path";
import { hostname as getHostname } from "node:os";
import { randomUUID } from "node:crypto";
import { monitorEventLoopDelay } from "node:perf_hooks";
@@ -50,6 +49,7 @@ import {
type PushNotifications,
type PushNotificationSender,
} from "./push/index.js";
import type { PushTokenStore } from "./push/token-store.js";
import type { ScriptHealthState } from "./script-health-monitor.js";
import type { ServiceProxySubsystem } from "./service-proxy.js";
import type { WorkspaceScriptRuntimeStore } from "./workspace-script-runtime-store.js";
@@ -585,6 +585,7 @@ export class VoiceAssistantWebSocketServer {
agentManager: AgentManager,
agentStorage: AgentStorage,
downloadTokenStore: DownloadTokenStore,
pushTokenStore: PushTokenStore,
paseoHome: string,
daemonConfigStore: DaemonConfigStore,
mcpBaseUrl: string | null,
@@ -688,7 +689,7 @@ export class VoiceAssistantWebSocketServer {
const pushLogger = this.logger.child({ module: "push" });
this.pushNotifications = createPushNotifications({
logger: pushLogger,
filePath: join(paseoHome, "push-tokens.json"),
store: pushTokenStore,
});
this.pushNotificationSender = pushNotificationSender ?? this.pushNotifications;