diff --git a/docs/cli.md b/docs/cli.md index 165be36e..2e076040 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -273,7 +273,7 @@ are stored under `.lousho/` in the current directory. See ```bash npx lousho observe start # collector on http://127.0.0.1:7777 npx lousho observe status # health + fleet counts -npx lousho observe stop # pidfile + SIGTERM +npx lousho observe stop # .pid + SIGTERM (checks the daemon serves --db) npx lousho observe rebuild-rollups # recompute insight aggregates npx lousho observe token create --project myapp # bearer key for non-loopback auth ``` @@ -283,7 +283,13 @@ set: traces, fleet monitoring, eval scores, the prompt registry and the context hub in one SQLite file (default `~/.lousho/observe.db`). The Studio Observe tabs read it. `start` takes `--port` (or `LOUSHO_OBSERVE_PORT`), `--db`, `--host`, `--require-token` and a repeatable `--webhook` for alert -hooks. See [The observe daemon](./observability.md#the-observe-daemon). +hooks. Each daemon writes its own `.pid` beside the db +(`observe.db` gives `observe.pid`), so two dbs in one folder never share a +pidfile. `start` exits 1 when the port already serves a different db, and +`stop` refuses to signal when the daemon at the pidfile's port reports another +db, and falls back to signalling when it cannot verify (unreachable, or an old +one-line pidfile). See +[The observe daemon](./observability.md#the-observe-daemon). ## `lousho prompt` diff --git a/llms-full.txt b/llms-full.txt index 2a367a48..788cd04e 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -5150,7 +5150,7 @@ are stored under `.lousho/` in the current directory. See ```bash npx lousho observe start # collector on http://127.0.0.1:7777 npx lousho observe status # health + fleet counts -npx lousho observe stop # pidfile + SIGTERM +npx lousho observe stop # .pid + SIGTERM (checks the daemon serves --db) npx lousho observe rebuild-rollups # recompute insight aggregates npx lousho observe token create --project myapp # bearer key for non-loopback auth ``` @@ -5160,7 +5160,13 @@ set: traces, fleet monitoring, eval scores, the prompt registry and the context hub in one SQLite file (default `~/.lousho/observe.db`). The Studio Observe tabs read it. `start` takes `--port` (or `LOUSHO_OBSERVE_PORT`), `--db`, `--host`, `--require-token` and a repeatable `--webhook` for alert -hooks. See [The observe daemon](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/observability.md#the-observe-daemon). +hooks. Each daemon writes its own `.pid` beside the db +(`observe.db` gives `observe.pid`), so two dbs in one folder never share a +pidfile. `start` exits 1 when the port already serves a different db, and +`stop` refuses to signal when the daemon at the pidfile's port reports another +db, and falls back to signalling when it cannot verify (unreachable, or an old +one-line pidfile). See +[The observe daemon](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/observability.md#the-observe-daemon). ## `lousho prompt` diff --git a/src/cli/observe.test.ts b/src/cli/observe.test.ts index 911b137a..52c6381a 100644 --- a/src/cli/observe.test.ts +++ b/src/cli/observe.test.ts @@ -64,7 +64,8 @@ describe('lousho observe', () => { }); it('start on the daemon\'s own port reports "already running" and exits 0', async () => { - const daemon = await start(); + const daemon = await startObserveDaemon({ dbPath: dbPath(), port: 0 }); + daemons.push(daemon); expect(await runObserve(['start', '--port', String(daemon.port), '--db', dbPath()])).toBe(0); expect(logs.at(-1)).toContain('already running'); }); @@ -93,6 +94,38 @@ describe('lousho observe', () => { expect(fs.existsSync(pidPath)).toBe(false); }); + it('start --db b.db against a daemon serving a.db exits 1 naming the other db', async () => { + const a = path.join(cwd, 'a.db'); + const daemon = await startObserveDaemon({ dbPath: a, port: 0 }); + daemons.push(daemon); + expect(await runObserve(['start', '--port', String(daemon.port), '--db', path.join(cwd, 'b.db')])).toBe(1); + expect(errors.at(-1)).toContain(`port ${daemon.port} is serving ${path.resolve(a)}`); + // the same db is still "already running" + expect(await runObserve(['start', '--port', String(daemon.port), '--db', a])).toBe(0); + expect(logs.at(-1)).toContain('already running'); + }); + + it('stop --db a.db does not signal a pid whose daemon serves b.db', async () => { + const b = await startObserveDaemon({ dbPath: path.join(cwd, 'b.db'), port: 0 }); + daemons.push(b); + const a = path.join(cwd, 'a.db'); + fs.writeFileSync(defaultPidPath(a), `${process.pid}\n${b.port}\n`); + const kill = vi.spyOn(process, 'kill').mockImplementation(() => true); + expect(await runObserve(['stop', '--db', a])).toBe(1); + expect(kill).not.toHaveBeenCalled(); + expect(errors.at(-1)).toContain('serves'); + expect(fs.existsSync(defaultPidPath(a))).toBe(true); + }); + + it('stop signals the pid when its daemon serves this db', async () => { + const a = path.join(cwd, 'a.db'); + const d = await startObserveDaemon({ dbPath: a, port: 0, pidPath: defaultPidPath(a) }); + daemons.push(d); + const kill = vi.spyOn(process, 'kill').mockImplementation(() => true); + expect(await runObserve(['stop', '--db', a])).toBe(0); + expect(kill).toHaveBeenCalledWith(process.pid, 'SIGTERM'); + }); + it('rebuild-rollups rebuilds the buckets of a local db', async () => { expect(await runObserve(['rebuild-rollups', '--db', dbPath()])).toBe(0); expect(logs.at(-1)).toContain('rebuilt'); diff --git a/src/cli/observe.ts b/src/cli/observe.ts index 78b3e0b5..61446e30 100644 --- a/src/cli/observe.ts +++ b/src/cli/observe.ts @@ -21,6 +21,7 @@ import { defaultDbPath, defaultPidPath, isObserveDaemon, + sameDbPath, probeHealth, startObserveDaemon, type HealthProbe, @@ -114,6 +115,25 @@ function envPort(): number { /* ------------------------------------------------------------------- start */ +/** `start` hit EADDRINUSE: say who owns the port. 0 = this db's daemon already runs; 1 = anything else. */ +async function reportPortInUse(port: number, host: string, dbPath: string): Promise { + const health = await probeHealth(port, host); + if (!isObserveDaemon(health)) { + console.error( + `lousho observe: port ${port} is taken by another service (its /health is not ${OBSERVE_SERVICE}). Pass --port , or stop that service.` + ); + return 1; + } + if (health.db !== undefined && !sameDbPath(health.db, dbPath)) { + console.error( + `lousho observe: port ${port} is serving ${health.db}, not ${dbPath}. Pass --port for this db, or stop that daemon first.` + ); + return 1; + } + console.log(`lousho observe: already running on http://${host}:${port} (service ${OBSERVE_SERVICE}, version ${health.version ?? 'unknown'}).`); + return 0; +} + async function runStart(args: string[]): Promise { const spec = SUB_SPECS.start; const { values, help } = parseCommand(spec, args); @@ -132,17 +152,7 @@ async function runStart(args: string[]): Promise { try { daemon = await startObserveDaemon({ dbPath, port, host, requireToken, pidPath, webhooks }); } catch (error) { - if ((error as NodeJS.ErrnoException).code === 'EADDRINUSE') { - const health = await probeHealth(port, host); - if (isObserveDaemon(health)) { - console.log(`lousho observe: already running on http://${host}:${port} (service ${OBSERVE_SERVICE}, version ${health.version ?? 'unknown'}).`); - return 0; - } - console.error( - `lousho observe: port ${port} is taken by another service (its /health is not ${OBSERVE_SERVICE}). Pass --port , or stop that service.` - ); - return 1; - } + if ((error as NodeJS.ErrnoException).code === 'EADDRINUSE') return reportPortInUse(port, host, dbPath); throw error; } @@ -215,6 +225,36 @@ async function runStatus(args: string[]): Promise { /* -------------------------------------------------------------------- stop */ +/** The pidfile's pid (line 1) and, for newer daemons, the port it listens on (line 2). */ +function readPidfile(pidPath: string): { pid?: number; port?: number } { + let text: string; + try { + text = fs.readFileSync(pidPath, 'utf8'); + } catch { + return {}; + } + const [pidLine = '', portLine] = text.split(/\r?\n/); + const pid = Number(pidLine.trim()); + const port = Number(portLine?.trim()); + return { + ...(Number.isInteger(pid) && pid > 0 ? { pid } : {}), + ...(Number.isInteger(port) && port > 0 ? { port } : {}), + }; +} + +/** + * What the daemon at the pidfile's port serves when that is not `dbPath` + * (a different db, or another service); `undefined` when it matches or cannot + * be verified (no port recorded, nothing answers). + */ +async function stopMismatch(port: number | undefined, dbPath: string): Promise { + if (port === undefined) return undefined; + const health = await probeHealth(port); + if (health === undefined) return undefined; + if (!isObserveDaemon(health)) return 'another service'; + return health.db !== undefined && !sameDbPath(health.db, dbPath) ? health.db : undefined; +} + async function runStop(args: string[]): Promise { const spec = SUB_SPECS.stop; const { values, help } = parseCommand(spec, args); @@ -224,18 +264,20 @@ async function runStop(args: string[]): Promise { } const dbPath = dbPathOf(values.db); const pidPath = defaultPidPath(dbPath); - let pid: number | undefined; - try { - pid = Number(fs.readFileSync(pidPath, 'utf8').trim()); - } catch { - pid = undefined; - } - if (pid === undefined || !Number.isInteger(pid) || pid <= 0) { + const { pid, port: pidPort } = readPidfile(pidPath); + if (pid === undefined) { console.error( `lousho observe: no pidfile at ${pidPath}. If a daemon is running, stop it with Ctrl+C (or 'kill ' / Task Manager).` ); return 1; } + const mismatch = await stopMismatch(pidPort, dbPath); + if (mismatch !== undefined) { + console.error( + `lousho observe: pid ${pid} in ${pidPath} serves ${mismatch}, not ${dbPath}; not signalling it. Remove the stale pidfile if it is wrong.` + ); + return 1; + } try { process.kill(pid, 'SIGTERM'); } catch (error) { diff --git a/src/observe/daemon.test.ts b/src/observe/daemon.test.ts index e6f0cd97..3f36bfb0 100644 --- a/src/observe/daemon.test.ts +++ b/src/observe/daemon.test.ts @@ -5,9 +5,10 @@ * port-collision detection. */ import { afterEach, describe, expect, it } from 'vitest'; -import { isObserveDaemon, probeHealth, startObserveDaemon, type ObserveDaemon } from './daemon'; +import * as nodePath from 'node:path'; +import { defaultPidPath, isObserveDaemon, probeHealth, sameDbPath, startObserveDaemon, type ObserveDaemon } from './daemon'; import { OBSERVE_SERVICE } from './envelope'; -import { createToken } from './store'; +import { createToken, openDb } from './store'; const T0 = 1_700_000_000_000; @@ -57,6 +58,24 @@ describe('GET /health', () => { }); }); +describe('per-db pidfile and db identity', () => { + it('names the pidfile after the db so two dbs in one folder never share it', () => { + expect(defaultPidPath('/d/a.db')).toBe(nodePath.resolve('/d/a.pid')); + expect(defaultPidPath('/d/b.db')).toBe(nodePath.resolve('/d/b.pid')); + expect(defaultPidPath('/d/observe.db')).toBe(nodePath.resolve('/d/observe.pid')); + expect(defaultPidPath('/d/data.sqlite')).toBe(nodePath.resolve('/d/data.sqlite.pid')); + }); + + it('/health reports the absolute db path; sameDbPath compares resolved paths', async () => { + const daemon = await start({ dbPath: 'mem-test.db', db: openDb(':memory:') }); + const health = await probeHealth(daemon.port); + expect(health?.db).toBe(nodePath.resolve('mem-test.db')); + expect(sameDbPath(health?.db, 'mem-test.db')).toBe(true); + expect(sameDbPath(health?.db, 'other.db')).toBe(false); + expect(sameDbPath(undefined, 'other.db')).toBe(false); + }); +}); + describe('POST /api/v1/ingest', () => { it('accepts a batch and serves it back through the read endpoints', async () => { const daemon = await start(); diff --git a/src/observe/daemon.ts b/src/observe/daemon.ts index 794c8722..2b563e7c 100644 --- a/src/observe/daemon.ts +++ b/src/observe/daemon.ts @@ -34,9 +34,23 @@ export function defaultDbPath(homeDir: string = os.homedir()): string { return path.join(homeDir, '.lousho', 'observe.db'); } -/** Where the pidfile goes for a given db path (same directory, fixed name). */ +/** + * Where the pidfile goes for a given db path: `/.pid`, + * so two daemons whose dbs share a folder never share a pidfile + * (`observe.db` keeps `observe.pid`). + */ export function defaultPidPath(dbPath: string): string { - return path.join(path.dirname(resolveForFs(dbPath)), 'observe.pid'); + const resolved = resolveForFs(dbPath); + const base = path.basename(resolved); + const stem = base.toLowerCase().endsWith('.db') && base.length > 3 ? base.slice(0, -3) : base; + return path.join(path.dirname(resolved), `${stem}.pid`); +} + +/** Whether two db paths name the same file (resolved; case-insensitive on win32). */ +export function sameDbPath(a: string | undefined, b: string): boolean { + if (a === undefined) return false; + const norm = (p: string): string => (process.platform === 'win32' ? resolveForFs(p).toLowerCase() : resolveForFs(p)); + return norm(a) === norm(b); } function resolveForFs(p: string): string { @@ -154,6 +168,7 @@ export async function startObserveDaemon(options: ObserveDaemonOptions = {}): Pr let bodyRead: Promise | undefined; const ctx: RouteContext = { db, + dbPath, req, res, params: match.params, @@ -205,7 +220,8 @@ export async function startObserveDaemon(options: ObserveDaemonOptions = {}): Pr const pidPath = options.pidPath; if (pidPath !== undefined) { fs.mkdirSync(path.dirname(pidPath), { recursive: true }); - fs.writeFileSync(pidPath, String(process.pid), 'utf8'); + // Line 2 is the port, so `stop` can ask this daemon which db it serves. + fs.writeFileSync(pidPath, `${process.pid}\n${boundPort}\n`, 'utf8'); } // The monitoring sweep: rewrites agents.status from heartbeat freshness, @@ -250,6 +266,8 @@ export interface HealthProbe { service: string; ok: boolean; version?: string; + /** Absolute path of the db the daemon serves. */ + db?: string; } /** diff --git a/src/observe/routes/httpKit.ts b/src/observe/routes/httpKit.ts index e598fb70..d76dc0ff 100644 --- a/src/observe/routes/httpKit.ts +++ b/src/observe/routes/httpKit.ts @@ -17,6 +17,8 @@ export { sendJson }; /** What a handler gets: the request, the matched `:params`, the query, the db, and a bounded body reader. */ export interface RouteContext { db: SqlDatabase; + /** The db file path the daemon was started with (`:memory:` for tests). */ + dbPath: string; req: http.IncomingMessage; res: http.ServerResponse; params: Record; diff --git a/src/observe/routes/scores.test.ts b/src/observe/routes/scores.test.ts index 4079d535..5fb003a8 100644 --- a/src/observe/routes/scores.test.ts +++ b/src/observe/routes/scores.test.ts @@ -45,6 +45,7 @@ function get(database: SqlDatabase, query: string): Captured { if (!matched) throw new Error('scores route not registered'); matched.route.handle({ db: database, + dbPath: ':memory:', req: {} as http.IncomingMessage, res, params: matched.params, diff --git a/src/observe/routes/status.ts b/src/observe/routes/status.ts index 0195d801..32a84df8 100644 --- a/src/observe/routes/status.ts +++ b/src/observe/routes/status.ts @@ -6,6 +6,7 @@ * row-count surface `lousho observe status` prints. */ +import * as path from 'node:path'; import { VERSION } from '../../version'; import { observeCounts } from '../store'; import { sendJson, type Route } from './httpKit'; @@ -14,8 +15,9 @@ export const statusRoutes: readonly Route[] = [ { method: 'GET', path: '/health', - handle: ({ res }) => { - sendJson(res, 200, { service: 'lousho-observe', ok: true, version: VERSION }); + handle: ({ res, dbPath }) => { + // `db` lets `start`/`stop` tell which database a port or pid serves. + sendJson(res, 200, { service: 'lousho-observe', ok: true, version: VERSION, db: dbPath === ':memory:' ? dbPath : path.resolve(dbPath) }); }, }, {