diff --git a/apps/api/src/local-data-root.ts b/apps/api/src/local-data-root.ts index 68c3959..33d9f22 100644 --- a/apps/api/src/local-data-root.ts +++ b/apps/api/src/local-data-root.ts @@ -69,6 +69,13 @@ interface InstanceConfiguration { schema_version: 1; } +export function readConfiguredLocalDataRoot(configFile = defaultInstanceConfigPath()) { + const configuration = JSON.parse(readFileSync(configFile, "utf8")) as Record; + const candidate = configuration.data_root ?? configuration.local_data_root; + if (typeof candidate !== "string" || !isAbsolute(candidate)) throw new Error("data_root_configuration_invalid"); + return resolve(candidate); +} + export interface ValidatedReadOnlyAssetRoot { absolute_root: string; ok: true; diff --git a/apps/api/src/main.ts b/apps/api/src/main.ts index 24dbac5..416f9bc 100644 --- a/apps/api/src/main.ts +++ b/apps/api/src/main.ts @@ -1,7 +1,10 @@ -import { resolve } from "node:path"; +import { join, resolve } from "node:path"; import { createApp } from "./app.js"; import { readBrowserSupportRelease } from "./browser-support.js"; +import { readConfiguredLocalDataRoot } from "./local-data-root.js"; +import { ManagedStorage } from "./managed-storage.js"; +import { StructuredJsonlLogger } from "./structured-log.js"; import { attachApiSupervisorControl, initializeApiCredentialClients, receiveApiCredentials } from "./supervisor-channel.js"; const credentialChannelEnabled = process.argv.includes("--dada-credential-stdin"); @@ -21,5 +24,27 @@ const controlPipeIndex = process.argv.indexOf("--dada-control-pipe"); if (controlPipeIndex >= 0) { const controlPipe = process.argv[controlPipeIndex + 1]; if (!controlPipe) throw new Error("Supervisor control pipe name is required."); - attachApiSupervisorControl(controlPipe, () => app.close()); + let storage: ManagedStorage | undefined; + const control = attachApiSupervisorControl(controlPipe, async () => { + await app.close(); + storage?.close(); + }); + try { + const dataRoot = readConfiguredLocalDataRoot(); + storage = new ManagedStorage({ dataRoot, databasePath: join(dataRoot, "db", "dada.sqlite3") }); + const logger = new StructuredJsonlLogger({ + component: "api", + directory: join(dataRoot, "logs", "api"), + onWriteFailure: () => { + try { + storage?.setLogAvailability(false); + } finally { + control.reportStatus("storage_unavailable"); + } + }, + }); + logger.write({ error_category: "none", status_category: "ready" }); + } catch { + control.reportStatus("storage_unavailable"); + } } diff --git a/apps/api/src/managed-storage.ts b/apps/api/src/managed-storage.ts index 6dd08b7..58af953 100644 --- a/apps/api/src/managed-storage.ts +++ b/apps/api/src/managed-storage.ts @@ -134,6 +134,7 @@ export class ManagedStorage { private readonly database: BetterSqlite3.Database; private dataRootWritable = true; private diskSpaceAvailable = true; + private logWritable = true; private sqliteWritable = true; private measurementBaselineBytes = 0; private recoveryRequiresRemeasure = false; @@ -241,7 +242,7 @@ export class ManagedStorage { getState(): StorageStateRow { const state = this.database.prepare("SELECT * FROM local_backend_storage_state WHERE singleton = 1").get() as Omit; const withReservations = { ...state, active_storage_reservations_bytes: this.activeReservationBytes() }; - if (!this.dataRootWritable || !this.diskSpaceAvailable || !this.sqliteWritable) { + if (!this.dataRootWritable || !this.diskSpaceAvailable || !this.sqliteWritable || !this.logWritable) { return { ...withReservations, storage_status: "unavailable" }; } return withReservations; @@ -260,7 +261,7 @@ export class ManagedStorage { if (!this.sqliteWritable) return; const state = this.database.prepare("SELECT managed_content_bytes FROM local_backend_storage_state WHERE singleton = 1").get() as { managed_content_bytes: number }; const classification = classifyCapacity(state.managed_content_bytes, this.activeReservationBytes()); - const storageStatus = !this.dataRootWritable || !this.diskSpaceAvailable || this.recoveryRequiresRemeasure + const storageStatus = !this.dataRootWritable || !this.diskSpaceAvailable || !this.logWritable || this.recoveryRequiresRemeasure ? "unavailable" : classification.storage_status; this.database.prepare(` @@ -291,6 +292,12 @@ export class ManagedStorage { this.refreshState(); } + setLogAvailability(writable: boolean) { + this.logWritable = writable; + if (!writable) this.recoveryRequiresRemeasure = true; + this.refreshState(); + } + private physicalManagedBytes() { return ["content", "managed-assets", "derived-assets"] .flatMap((root) => listFiles(resolvePathWithinRoot(this.dataRoot, root))) diff --git a/apps/api/src/structured-log.ts b/apps/api/src/structured-log.ts new file mode 100644 index 0000000..47e606b --- /dev/null +++ b/apps/api/src/structured-log.ts @@ -0,0 +1,165 @@ +import { + appendFileSync, + existsSync, + mkdirSync, + readdirSync, + renameSync, + rmSync, + statSync, +} from "node:fs"; +import { join } from "node:path"; + +export const LOG_SIZE_LIMIT_BYTES = 10 * 1024 * 1024; +export const LOG_FILE_LIMIT = 10; +export const LOG_RETENTION_DAYS = 30; + +const statusCategories = new Set([ + "starting", "ready", "completed", "failed", "unavailable", "degraded", + "blocked", "stopping", "stopped", +]); +const errorCategories = new Set([ + "none", "invalid_input", "storage_unavailable", "log_write_failed", + "service_unavailable", "timeout", "internal_error", +]); +const identifierPattern = /^[a-zA-Z0-9][a-zA-Z0-9_-]{0,63}$/; +const codePattern = /^[a-z][a-z0-9_]{0,63}$/; + +export interface StructuredLogInput { + correlation_id?: string; + duration_ms?: number; + error_category?: string; + object_id?: string; + status_category: string; +} + +export interface StructuredJsonlLoggerOptions { + component: "api"; + directory: string; + now?: () => Date; + onWriteFailure?: () => void; +} + +export class LogWriteError extends Error { + constructor(options?: ErrorOptions) { + super("log_write_failed", options); + } +} + +export class StructuredJsonlLogger { + private readonly component: "api"; + private readonly directory: string; + private readonly now: () => Date; + private readonly onWriteFailure: () => void; + private initialized = false; + private currentSize = 0; + + constructor(options: StructuredJsonlLoggerOptions) { + this.component = options.component; + this.directory = options.directory; + this.now = options.now ?? (() => new Date()); + this.onWriteFailure = options.onWriteFailure ?? (() => undefined); + } + + write(input: StructuredLogInput | Record) { + try { + this.initialize(); + const line = `${JSON.stringify(sanitizeLogEntry(this.component, input, this.now()))}\n`; + const bytes = Buffer.byteLength(line); + if (bytes > LOG_SIZE_LIMIT_BYTES) throw new Error("log_entry_too_large"); + if (this.currentSize > 0 && this.currentSize + bytes > LOG_SIZE_LIMIT_BYTES) this.rotate(); + appendFileSync(this.activePath(), line, { encoding: "utf8" }); + this.currentSize += bytes; + } catch (error) { + if (error instanceof LogWriteError) throw error; + this.fail(error); + } + } + + maintain() { + try { + mkdirSync(this.directory, { recursive: true }); + this.maintainFiles(); + this.currentSize = existsSync(this.activePath()) ? statSync(this.activePath()).size : 0; + this.initialized = true; + } catch (error) { + if (error instanceof LogWriteError) throw error; + this.fail(error); + } + } + + private initialize() { + if (!this.initialized) this.maintain(); + } + + private maintainFiles() { + const cutoff = this.now().getTime() - LOG_RETENTION_DAYS * 24 * 60 * 60 * 1000; + for (const file of this.files()) { + const path = join(this.directory, file.name); + if (file.index > LOG_FILE_LIMIT - 1 || statSync(path).mtimeMs < cutoff) rmSync(path, { force: true }); + } + const remaining = this.files(); + for (const file of remaining.slice(LOG_FILE_LIMIT)) rmSync(join(this.directory, file.name), { force: true }); + } + + private rotate() { + const last = join(this.directory, `${this.component}.${LOG_FILE_LIMIT - 1}.jsonl`); + rmSync(last, { force: true }); + for (let index = LOG_FILE_LIMIT - 2; index >= 1; index -= 1) { + const source = join(this.directory, `${this.component}.${index}.jsonl`); + if (existsSync(source)) renameSync(source, join(this.directory, `${this.component}.${index + 1}.jsonl`)); + } + if (existsSync(this.activePath())) renameSync(this.activePath(), join(this.directory, `${this.component}.1.jsonl`)); + this.currentSize = 0; + } + + private files() { + if (!existsSync(this.directory)) return []; + const pattern = new RegExp(`^${this.component}(?:\\.(\\d+))?\\.jsonl$`); + return readdirSync(this.directory) + .map((name) => ({ match: pattern.exec(name), name })) + .filter((entry): entry is { match: RegExpExecArray; name: string } => entry.match !== null) + .map(({ match, name }) => ({ index: match[1] === undefined ? 0 : Number(match[1]), name })) + .sort((left, right) => left.index - right.index); + } + + private activePath() { + return join(this.directory, `${this.component}.jsonl`); + } + + private fail(error: unknown): never { + try { + this.onWriteFailure(); + } catch { + // State persistence failure must not replace the stable redacted error. + } + throw new LogWriteError({ cause: error }); + } +} + +export function sanitizeLogEntry(component: "api", input: StructuredLogInput | Record, now: Date) { + const value = input as Record; + const entry: Record = { + component, + schema_version: "1.0", + status_category: statusCategories.has(String(value.status_category)) ? value.status_category : "failed", + timestamp: now.toISOString(), + }; + if (typeof value.correlation_id === "string" && identifierPattern.test(value.correlation_id)) entry.correlation_id = value.correlation_id; + if (typeof value.object_id === "string" && identifierPattern.test(value.object_id)) entry.object_id = value.object_id; + if (typeof value.duration_ms === "number" && Number.isSafeInteger(value.duration_ms) && value.duration_ms >= 0) entry.duration_ms = value.duration_ms; + if (typeof value.error_category === "string") { + entry.error_category = errorCategories.has(value.error_category) ? value.error_category : "internal_error"; + } + return entry; +} + +export function sanitizeDiagnosticRecord(input: Record) { + return { + app_version: typeof input.app_version === "string" && /^[0-9]+(?:\.[0-9]+){2,3}$/.test(input.app_version) ? input.app_version : "0.0.0", + checked_at: typeof input.checked_at === "string" && !Number.isNaN(Date.parse(input.checked_at)) ? input.checked_at : new Date(0).toISOString(), + component: input.component === "api" || input.component === "worker" || input.component === "supervisor" ? input.component : "supervisor", + message_key: typeof input.message_key === "string" && codePattern.test(input.message_key) ? input.message_key : "diagnostic_redacted", + result: input.result === "pass" || input.result === "warning" || input.result === "fail" ? input.result : "fail", + stable_check_code: typeof input.stable_check_code === "string" && codePattern.test(input.stable_check_code) ? input.stable_check_code : "invalid_check", + }; +} diff --git a/apps/api/src/supervisor-channel.ts b/apps/api/src/supervisor-channel.ts index 41a68b3..45f41f4 100644 --- a/apps/api/src/supervisor-channel.ts +++ b/apps/api/src/supervisor-channel.ts @@ -33,8 +33,14 @@ export function initializeApiCredentialClients(credentials: Record<(typeof API_C export function attachApiSupervisorControl(pipeName: string, shutdown: () => Promise) { const socket = createConnection(`\\\\.\\pipe\\${pipeName}`); let pending = ""; + let connected = false; + const pendingStatuses: string[] = []; socket.setEncoding("utf8"); - socket.on("connect", () => socket.write("ready\n")); + socket.on("connect", () => { + connected = true; + socket.write("ready\n"); + for (const status of pendingStatuses.splice(0)) socket.write(`${status}\n`); + }); socket.on("data", (chunk) => { pending += chunk; if (!pending.includes("\n")) return; @@ -42,5 +48,11 @@ export function attachApiSupervisorControl(pipeName: string, shutdown: () => Pro pending = ""; if (command === "shutdown") void shutdown().finally(() => socket.end()); }); - return socket; + return { + reportStatus(status: "storage_unavailable") { + if (socket.destroyed) return; + if (connected) socket.write(`${status}\n`); else pendingStatuses.push(status); + }, + socket, + }; } diff --git a/apps/worker/src/ai-call-gate.ts b/apps/worker/src/ai-call-gate.ts new file mode 100644 index 0000000..741fc58 --- /dev/null +++ b/apps/worker/src/ai-call-gate.ts @@ -0,0 +1,47 @@ +import type { StructuredJsonlLogger } from "./structured-log.js"; + +export class WorkerAiCallGate { + private readonly getStorageStatus: () => "active" | "full" | "unavailable"; + private readonly logger: StructuredJsonlLogger; + + constructor(input: { + getStorageStatus: () => "active" | "full" | "unavailable"; + logger: StructuredJsonlLogger; + }) { + this.getStorageStatus = input.getStorageStatus; + this.logger = input.logger; + } + + async execute(correlationId: string, objectId: string, externalCall: () => Promise) { + const initialStatus = this.getStorageStatus(); + if (initialStatus === "full") return { reason: "storage_full" as const, status: "blocked" as const }; + if (initialStatus === "unavailable") return { reason: "storage_unavailable" as const, status: "blocked" as const }; + try { + this.logger.write({ + correlation_id: correlationId, + error_category: "none", + object_id: objectId, + status_category: "starting", + }); + } catch { + return { reason: "storage_unavailable" as const, status: "blocked" as const }; + } + const statusAfterLog = this.getStorageStatus(); + if (statusAfterLog === "full") return { reason: "storage_full" as const, status: "blocked" as const }; + if (statusAfterLog === "unavailable") return { reason: "storage_unavailable" as const, status: "blocked" as const }; + const started = Date.now(); + const value = await externalCall(); + try { + this.logger.write({ + correlation_id: correlationId, + duration_ms: Date.now() - started, + error_category: "none", + object_id: objectId, + status_category: "completed", + }); + } catch { + // The completed call cannot be undone; the write failure blocks every later call. + } + return { status: "completed" as const, value }; + } +} diff --git a/apps/worker/src/runtime-config.ts b/apps/worker/src/runtime-config.ts new file mode 100644 index 0000000..4a4abcd --- /dev/null +++ b/apps/worker/src/runtime-config.ts @@ -0,0 +1,12 @@ +import { readFileSync } from "node:fs"; +import { isAbsolute, join, resolve } from "node:path"; + +export function readConfiguredLocalDataRoot(environment: NodeJS.ProcessEnv = process.env) { + const localAppData = environment.LOCALAPPDATA; + if (!localAppData || !isAbsolute(localAppData)) throw new Error("local_app_data_unavailable"); + const configFile = join(localAppData, "Dada", "P0A", "config", "instance.json"); + const configuration = JSON.parse(readFileSync(configFile, "utf8")) as Record; + const candidate = configuration.data_root ?? configuration.local_data_root; + if (typeof candidate !== "string" || !isAbsolute(candidate)) throw new Error("data_root_configuration_invalid"); + return resolve(candidate); +} diff --git a/apps/worker/src/storage-status.ts b/apps/worker/src/storage-status.ts new file mode 100644 index 0000000..b8d0444 --- /dev/null +++ b/apps/worker/src/storage-status.ts @@ -0,0 +1,32 @@ +import { createRequire } from "node:module"; + +import type BetterSqlite3 from "better-sqlite3"; + +const require = createRequire(import.meta.url); +const Database = require("better-sqlite3") as typeof BetterSqlite3; + +export class WorkerStorageStatus { + private readonly database: BetterSqlite3.Database; + + constructor(databasePath: string) { + this.database = new Database(databasePath); + this.database.pragma("busy_timeout = 5000"); + } + + getStatus(): "active" | "full" | "unavailable" { + const row = this.database.prepare("SELECT storage_status FROM local_backend_storage_state WHERE singleton = 1").get() as { storage_status: "active" | "full" | "unavailable" }; + return row.storage_status; + } + + markLogUnavailable() { + this.database.prepare(` + UPDATE local_backend_storage_state + SET storage_status = 'unavailable', measured_at = ?, version = version + 1 + WHERE singleton = 1 + `).run(new Date().toISOString()); + } + + close() { + this.database.close(); + } +} diff --git a/apps/worker/src/structured-log.ts b/apps/worker/src/structured-log.ts new file mode 100644 index 0000000..290bb3e --- /dev/null +++ b/apps/worker/src/structured-log.ts @@ -0,0 +1,129 @@ +import { + appendFileSync, + existsSync, + mkdirSync, + readdirSync, + renameSync, + rmSync, + statSync, +} from "node:fs"; +import { join } from "node:path"; + +export const LOG_SIZE_LIMIT_BYTES = 10 * 1024 * 1024; +export const LOG_FILE_LIMIT = 10; +export const LOG_RETENTION_DAYS = 30; + +const statusCategories = new Set(["starting", "ready", "completed", "failed", "unavailable", "degraded", "blocked", "stopping", "stopped"]); +const errorCategories = new Set(["none", "invalid_input", "storage_unavailable", "log_write_failed", "service_unavailable", "timeout", "internal_error"]); +const identifierPattern = /^[a-zA-Z0-9][a-zA-Z0-9_-]{0,63}$/; + +export interface StructuredLogInput { + correlation_id?: string; + duration_ms?: number; + error_category?: string; + object_id?: string; + status_category: string; +} + +export class LogWriteError extends Error { + constructor(options?: ErrorOptions) { + super("log_write_failed", options); + } +} + +export class StructuredJsonlLogger { + private readonly directory: string; + private readonly now: () => Date; + private readonly onWriteFailure: () => void; + private initialized = false; + private currentSize = 0; + + constructor(options: { component: "worker"; directory: string; now?: () => Date; onWriteFailure?: () => void }) { + this.directory = options.directory; + this.now = options.now ?? (() => new Date()); + this.onWriteFailure = options.onWriteFailure ?? (() => undefined); + } + + write(input: StructuredLogInput | Record) { + try { + this.initialize(); + const line = `${JSON.stringify(sanitizeWorkerLogEntry(input, this.now()))}\n`; + const bytes = Buffer.byteLength(line); + if (bytes > LOG_SIZE_LIMIT_BYTES) throw new Error("log_entry_too_large"); + if (this.currentSize > 0 && this.currentSize + bytes > LOG_SIZE_LIMIT_BYTES) this.rotate(); + appendFileSync(this.activePath(), line, { encoding: "utf8" }); + this.currentSize += bytes; + } catch (error) { + if (error instanceof LogWriteError) throw error; + this.fail(error); + } + } + + maintain() { + try { + mkdirSync(this.directory, { recursive: true }); + const cutoff = this.now().getTime() - LOG_RETENTION_DAYS * 24 * 60 * 60 * 1000; + for (const file of this.files()) { + const path = join(this.directory, file.name); + if (file.index > LOG_FILE_LIMIT - 1 || statSync(path).mtimeMs < cutoff) rmSync(path, { force: true }); + } + for (const file of this.files().slice(LOG_FILE_LIMIT)) rmSync(join(this.directory, file.name), { force: true }); + this.currentSize = existsSync(this.activePath()) ? statSync(this.activePath()).size : 0; + this.initialized = true; + } catch (error) { + if (error instanceof LogWriteError) throw error; + this.fail(error); + } + } + + private initialize() { + if (!this.initialized) this.maintain(); + } + + private rotate() { + rmSync(join(this.directory, `worker.${LOG_FILE_LIMIT - 1}.jsonl`), { force: true }); + for (let index = LOG_FILE_LIMIT - 2; index >= 1; index -= 1) { + const source = join(this.directory, `worker.${index}.jsonl`); + if (existsSync(source)) renameSync(source, join(this.directory, `worker.${index + 1}.jsonl`)); + } + if (existsSync(this.activePath())) renameSync(this.activePath(), join(this.directory, "worker.1.jsonl")); + this.currentSize = 0; + } + + private files() { + if (!existsSync(this.directory)) return []; + return readdirSync(this.directory) + .map((name) => ({ match: /^worker(?:\.(\d+))?\.jsonl$/.exec(name), name })) + .filter((entry): entry is { match: RegExpExecArray; name: string } => entry.match !== null) + .map(({ match, name }) => ({ index: match[1] === undefined ? 0 : Number(match[1]), name })) + .sort((left, right) => left.index - right.index); + } + + private activePath() { + return join(this.directory, "worker.jsonl"); + } + + private fail(error: unknown): never { + try { + this.onWriteFailure(); + } catch { + // State persistence failure must not replace the stable redacted error. + } + throw new LogWriteError({ cause: error }); + } +} + +function sanitizeWorkerLogEntry(input: StructuredLogInput | Record, now: Date) { + const value = input as Record; + const entry: Record = { + component: "worker", + schema_version: "1.0", + status_category: statusCategories.has(String(value.status_category)) ? value.status_category : "failed", + timestamp: now.toISOString(), + }; + if (typeof value.correlation_id === "string" && identifierPattern.test(value.correlation_id)) entry.correlation_id = value.correlation_id; + if (typeof value.object_id === "string" && identifierPattern.test(value.object_id)) entry.object_id = value.object_id; + if (typeof value.duration_ms === "number" && Number.isSafeInteger(value.duration_ms) && value.duration_ms >= 0) entry.duration_ms = value.duration_ms; + if (typeof value.error_category === "string") entry.error_category = errorCategories.has(value.error_category) ? value.error_category : "internal_error"; + return entry; +} diff --git a/apps/worker/src/supervisor-channel.ts b/apps/worker/src/supervisor-channel.ts index c06e1b2..dc6d17e 100644 --- a/apps/worker/src/supervisor-channel.ts +++ b/apps/worker/src/supervisor-channel.ts @@ -33,8 +33,14 @@ export function initializeWorkerCredentialClient(credentials: Record<(typeof WOR export function attachWorkerSupervisorControl(pipeName: string, shutdown: () => Promise | void) { const socket = createConnection(`\\\\.\\pipe\\${pipeName}`); let pending = ""; + let connected = false; + const pendingStatuses: string[] = []; socket.setEncoding("utf8"); - socket.on("connect", () => socket.write("ready\n")); + socket.on("connect", () => { + connected = true; + socket.write("ready\n"); + for (const status of pendingStatuses.splice(0)) socket.write(`${status}\n`); + }); socket.on("data", (chunk) => { pending += chunk; if (!pending.includes("\n")) return; @@ -42,5 +48,11 @@ export function attachWorkerSupervisorControl(pipeName: string, shutdown: () => pending = ""; if (command === "shutdown") void Promise.resolve(shutdown()).finally(() => socket.end()); }); - return socket; + return { + reportStatus(status: "storage_unavailable") { + if (socket.destroyed) return; + if (connected) socket.write(`${status}\n`); else pendingStatuses.push(status); + }, + socket, + }; } diff --git a/apps/worker/src/worker.ts b/apps/worker/src/worker.ts index 8a288f0..5144815 100644 --- a/apps/worker/src/worker.ts +++ b/apps/worker/src/worker.ts @@ -1,5 +1,10 @@ import { parentPort } from "node:worker_threads"; +import { join } from "node:path"; +import { WorkerAiCallGate } from "./ai-call-gate.js"; +import { readConfiguredLocalDataRoot } from "./runtime-config.js"; +import { StructuredJsonlLogger } from "./structured-log.js"; +import { WorkerStorageStatus } from "./storage-status.js"; import { attachWorkerSupervisorControl, initializeWorkerCredentialClient, receiveWorkerCredentials } from "./supervisor-channel.js"; export function handleWorkerProbe(message: unknown) { @@ -20,5 +25,31 @@ if (!workerPort && process.argv.includes("--dada-credential-stdin")) { const controlPipe = process.argv[controlPipeIndex + 1]; if (controlPipeIndex < 0 || !controlPipe) throw new Error("Supervisor control pipe name is required."); const keepAlive = setInterval(() => undefined, 30_000); - attachWorkerSupervisorControl(controlPipe, () => clearInterval(keepAlive)); + let storage: WorkerStorageStatus | undefined; + const control = attachWorkerSupervisorControl(controlPipe, () => { + clearInterval(keepAlive); + storage?.close(); + }); + let storageStatus: "active" | "unavailable" = "active"; + try { + const dataRoot = readConfiguredLocalDataRoot(); + storage = new WorkerStorageStatus(join(dataRoot, "db", "dada.sqlite3")); + const logger = new StructuredJsonlLogger({ + component: "worker", + directory: join(dataRoot, "logs", "worker"), + onWriteFailure: () => { + storageStatus = "unavailable"; + try { + storage?.markLogUnavailable(); + } finally { + control.reportStatus("storage_unavailable"); + } + }, + }); + logger.write({ error_category: "none", status_category: "ready" }); + new WorkerAiCallGate({ getStorageStatus: () => storageStatus === "unavailable" ? storageStatus : (storage?.getStatus() ?? "unavailable"), logger }); + } catch { + storageStatus = "unavailable"; + control.reportStatus("storage_unavailable"); + } } diff --git a/package.json b/package.json index 0204a36..d7d7f60 100644 --- a/package.json +++ b/package.json @@ -34,7 +34,9 @@ "test:wp0-06": "node scripts/run-wp0-06-validation.mjs", "test:wp0-06:red": "node scripts/run-wp0-06-validation.mjs --phase red", "test:wp0-07": "node scripts/run-wp0-07-validation.mjs", - "test:wp0-07:red": "node scripts/run-wp0-07-validation.mjs --phase red" + "test:wp0-07:red": "node scripts/run-wp0-07-validation.mjs --phase red", + "test:wp0-08": "node scripts/run-wp0-08-validation.mjs", + "test:wp0-08:red": "node scripts/run-wp0-08-validation.mjs --phase red" }, "devDependencies": { "@playwright/test": "1.62.0", diff --git a/scripts/run-wp0-08-validation.mjs b/scripts/run-wp0-08-validation.mjs new file mode 100644 index 0000000..1fbe274 --- /dev/null +++ b/scripts/run-wp0-08-validation.mjs @@ -0,0 +1,79 @@ +import { createHash } from "node:crypto"; +import { spawnSync } from "node:child_process"; +import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"; +import { resolve } from "node:path"; + +const phaseIndex = process.argv.indexOf("--phase"); +const phase = phaseIndex >= 0 ? process.argv[phaseIndex + 1] : "green"; +if (!["red", "green"].includes(phase)) throw new Error(`Unsupported phase: ${phase}`); +const runId = process.env.DADA_TDD_RUN_ID ?? `wp0-08-${phase}-${new Date().toISOString().replace(/[^0-9]/g, "")}`; +const runDirectory = resolve("artifacts", "tdd", runId); +const caseId = "TDD-WP0-LOG-001-rotation-redaction"; +const caseDirectory = resolve(runDirectory, "cases", caseId); +const uiDirectory = resolve(runDirectory, "ui", "system-ui", "unavailable"); +if (existsSync(runDirectory)) throw new Error(`Evidence run already exists: ${runId}`); +mkdirSync(caseDirectory, { recursive: true }); +mkdirSync(uiDirectory, { recursive: true }); + +const commandSpecs = phase === "red" + ? [ + ["pnpm exec vitest run tests/integration/wp0-08-structured-log.test.ts", ["exec", "vitest", "run", "tests/integration/wp0-08-structured-log.test.ts"]], + ["pnpm exec vitest run tests/worker/wp0-08-log-availability.test.ts", ["exec", "vitest", "run", "tests/worker/wp0-08-log-availability.test.ts"]], + ] + : [ + ["pnpm test:integration", ["test:integration"]], + ["pnpm test:worker", ["test:worker"]], + ["pnpm test:security", ["test:security"]], + ["pnpm test:package", ["test:package"]], + ["pnpm validate:tdd-trace", ["validate:tdd-trace"]], + ]; +const environment = { ...process.env, DADA_EVIDENCE_DIR_LOG: caseDirectory, DADA_EVIDENCE_DIR_SUP: uiDirectory }; +const startedAt = new Date().toISOString(); +const commands = []; +for (const [command, args] of commandSpecs) { + const started_at = new Date().toISOString(); + const executable = process.platform === "win32" ? (process.env.ComSpec ?? "cmd.exe") : "pnpm"; + const actualArgs = process.platform === "win32" ? ["/d", "/s", "/c", `pnpm ${args.join(" ")}`] : args; + const execution = spawnSync(executable, actualArgs, { encoding: "utf8", env: environment }); + if (execution.stdout) process.stdout.write(execution.stdout); + if (execution.stderr) process.stderr.write(execution.stderr); + commands.push({ command, exit_code: execution.status ?? 1, finished_at: new Date().toISOString(), started_at }); +} + +writeFileSync(resolve(caseDirectory, "commands.json"), `${JSON.stringify({ commands, phase, run_id: runId, schema_version: "1.0" }, null, 2)}\n`); +const evidenceRefs = ["log-manifest.json", "redaction.json", "response.json", "external-calls.json"]; +const missingEvidence = phase === "green" ? evidenceRefs.filter((path) => !existsSync(resolve(caseDirectory, path))) : []; +const uiEvidenceRefs = ["ui/system-ui/unavailable/screenshots/storage-unavailable.png"]; +const missingUiEvidence = phase === "green" ? uiEvidenceRefs.filter((path) => !existsSync(resolve(runDirectory, path))) : []; +const commandState = phase === "red" ? commands.every((item) => item.exit_code !== 0) : commands.every((item) => item.exit_code === 0); +const status = phase === "red" ? (commandState ? "red_confirmed" : "failed") : (commandState && missingEvidence.length === 0 && missingUiEvidence.length === 0 ? "passed" : "failed"); +const result = { + acceptance_criteria: ["AC-41", "AC-56"], + automation: ["automated"], + commit: spawnSync("git", ["rev-parse", "--short", "HEAD"], { encoding: "utf8" }).stdout.trim(), + environment: { arch: process.arch, node: process.version.slice(1), os: process.platform }, + evidence_refs: evidenceRefs, + finished_at: new Date().toISOString(), + layer: ["DB", "WORKER", "PACKAGE_SECURITY"], + manifest: { path: "tasks.manifest.json", sha256: createHash("sha256").update(readFileSync("tasks.manifest.json")).digest("hex").toUpperCase() }, + missing_evidence: missingEvidence, + missing_ui_evidence: missingUiEvidence, + parent_family: "TDD-WP0-LOG-001", + phase, + release_gate: ["work_package:WP-0", "release:P0-A"], + requirements: ["NFR-09", "PRIV-01", "PRIV-02"], + run_id: runId, + schema_version: "1.0", + started_at: startedAt, + status, + task_id: "TASK-WP0-08", + test_id: caseId, + ui_evidence_refs: uiEvidenceRefs, + work_package: "WP-0", + worktree_under_test: spawnSync("git", ["status", "--porcelain"], { encoding: "utf8" }).stdout.trim() ? "uncommitted implementation" : "clean committed implementation", +}; +writeFileSync(resolve(caseDirectory, "result.json"), `${JSON.stringify(result, null, 2)}\n`); +const summary = { cases: [{ missing_evidence: missingEvidence, status, test_id: caseId }], phase, run_id: runId, status }; +writeFileSync(resolve(runDirectory, "evidence.json"), `${JSON.stringify(summary, null, 2)}\n`); +console.log(JSON.stringify(summary, null, 2)); +if (status !== (phase === "red" ? "red_confirmed" : "passed")) process.exit(1); diff --git a/supervisor/Dada.Supervisor.Tests/Program.cs b/supervisor/Dada.Supervisor.Tests/Program.cs index 8b80f03..0851980 100644 --- a/supervisor/Dada.Supervisor.Tests/Program.cs +++ b/supervisor/Dada.Supervisor.Tests/Program.cs @@ -32,6 +32,7 @@ internal static class Program { var security = await TestCredentialBoundaryAsync(); var supervisor = await TestSupervisorLifecycleAsync(); + TestStructuredLogging(); WriteEvidence(Environment.GetEnvironmentVariable("DADA_EVIDENCE_DIR_SEC"), security); WriteEvidence(Environment.GetEnvironmentVariable("DADA_EVIDENCE_DIR_SUP"), supervisor); Console.WriteLine(JsonSerializer.Serialize(new { security = "passed", supervisor = "passed" }, JsonOptions)); @@ -44,6 +45,35 @@ internal static class Program } } + private static void TestStructuredLogging() + { + Equal(10 * 1024 * 1024L, StructuredJsonlLogger.SizeLimitBytes, "Supervisor log byte limit"); + Equal(10, StructuredJsonlLogger.FileLimit, "Supervisor log file limit"); + Equal(30, StructuredJsonlLogger.RetentionDays, "Supervisor log retention"); + var directory = Path.Combine(Path.GetTempPath(), $"dada-supervisor-log-{Guid.NewGuid():N}"); + try + { + var logger = new StructuredJsonlLogger(directory, "supervisor", sizeLimitBytes: 512, fileLimit: 10); + for (var index = 0; index < 40; index++) + { + logger.Write(new StructuredLogEvent("completed", $"corr_{index}", $"obj_{index}", index, "none")); + } + var files = Directory.GetFiles(directory, "*.jsonl"); + True(files.Length <= StructuredJsonlLogger.FileLimit, "Supervisor log file cap"); + True(files.All(path => new FileInfo(path).Length <= 512), "Supervisor log rotation byte boundary"); + foreach (var line in files.SelectMany(File.ReadLines)) + { + using var document = JsonDocument.Parse(line); + var keys = document.RootElement.EnumerateObject().Select(property => property.Name).ToHashSet(StringComparer.Ordinal); + True(keys.IsSubsetOf(["schema_version", "timestamp", "component", "status_category", "correlation_id", "object_id", "duration_ms", "error_category"]), "Supervisor log field allowlist"); + } + } + finally + { + if (Directory.Exists(directory)) Directory.Delete(directory, recursive: true); + } + } + private static async Task TestCredentialBoundaryAsync() { var store = new TestCredentialStore(); @@ -184,6 +214,11 @@ internal static class Program { EqualSequence(new[] { "打开 Dada", "运行状态", "打开诊断", "重新启动服务", "退出 Dada" }, menuForm.TrayMenuLabels, "tray menu order"); } + using (var diagnostics = new DiagnosticsForm(SupervisorState.StorageUnavailable)) + { + True(diagnostics.CopyPayload.Contains("log_write_failed", StringComparison.Ordinal), "diagnostic log status"); + False(diagnostics.CopyPayload.Contains("CredentialBlob", StringComparison.Ordinal), "diagnostic credential redaction"); + } var screenshotPath = Path.Combine(Environment.GetEnvironmentVariable("DADA_EVIDENCE_DIR_SUP") ?? Path.GetTempPath(), "screenshots", "system-ui.png"); Directory.CreateDirectory(Path.GetDirectoryName(screenshotPath)!); diff --git a/supervisor/Dada.Supervisor/Diagnostics.cs b/supervisor/Dada.Supervisor/Diagnostics.cs index 158ee22..aae58b3 100644 --- a/supervisor/Dada.Supervisor/Diagnostics.cs +++ b/supervisor/Dada.Supervisor/Diagnostics.cs @@ -9,9 +9,12 @@ internal enum DiagnosticResult internal sealed record DiagnosticCheck(string CheckCode, DiagnosticResult Result, string MessageKey, DateTimeOffset CheckedAt) { - internal static DiagnosticCheck Pass(string checkCode, string messageKey) => new(checkCode, DiagnosticResult.Pass, messageKey, DateTimeOffset.UtcNow); - internal static DiagnosticCheck Warning(string checkCode, string messageKey) => new(checkCode, DiagnosticResult.Warning, messageKey, DateTimeOffset.UtcNow); - internal static DiagnosticCheck Fail(string checkCode, string messageKey) => new(checkCode, DiagnosticResult.Fail, messageKey, DateTimeOffset.UtcNow); + internal static DiagnosticCheck Pass(string checkCode, string messageKey) => Create(checkCode, DiagnosticResult.Pass, messageKey); + internal static DiagnosticCheck Warning(string checkCode, string messageKey) => Create(checkCode, DiagnosticResult.Warning, messageKey); + internal static DiagnosticCheck Fail(string checkCode, string messageKey) => Create(checkCode, DiagnosticResult.Fail, messageKey); + + private static DiagnosticCheck Create(string checkCode, DiagnosticResult result, string messageKey) => + new(RedactionPolicy.Code(checkCode, "invalid_check"), result, RedactionPolicy.Code(messageKey, "diagnostic_redacted"), DateTimeOffset.UtcNow); } internal sealed record DiagnosticStorageSummary(long UsedBytes, long CapacityBytes, long ReclaimableBytes); diff --git a/supervisor/Dada.Supervisor/ManagedChildProcess.cs b/supervisor/Dada.Supervisor/ManagedChildProcess.cs index 953c42b..8f4a646 100644 --- a/supervisor/Dada.Supervisor/ManagedChildProcess.cs +++ b/supervisor/Dada.Supervisor/ManagedChildProcess.cs @@ -8,18 +8,32 @@ internal sealed class ManagedChildProcess : IAsyncDisposable { private readonly NamedPipeServerStream controlPipe; private readonly StreamWriter controlWriter; + private readonly StreamReader controlReader; + private Action? statusReceived; private bool stopping; - private ManagedChildProcess(Process process, NamedPipeServerStream controlPipe, StreamWriter controlWriter) + private ManagedChildProcess(Process process, NamedPipeServerStream controlPipe, StreamReader controlReader, StreamWriter controlWriter) { Process = process; this.controlPipe = controlPipe; + this.controlReader = controlReader; this.controlWriter = controlWriter; + _ = ListenForStatusAsync(); } internal Process Process { get; } internal bool IsStopping => stopping; internal static TimeSpan ShutdownDeadline { get; } = TimeSpan.FromSeconds(15); + internal string? LastStatus { get; private set; } + internal event Action StatusReceived + { + add + { + statusReceived += value; + if (LastStatus is not null) value(LastStatus); + } + remove => statusReceived -= value; + } internal static async Task StartAsync( ProcessStartInfo startInfo, @@ -48,7 +62,7 @@ internal sealed class ManagedChildProcess : IAsyncDisposable var ready = await reader.ReadLineAsync(timeout.Token); if (!string.Equals(ready, "ready", StringComparison.Ordinal)) throw new InvalidOperationException("Managed child did not report ready."); var writer = new StreamWriter(pipe, new UTF8Encoding(false), leaveOpen: true) { AutoFlush = true }; - return new ManagedChildProcess(process, pipe, writer); + return new ManagedChildProcess(process, pipe, reader, writer); } catch { @@ -59,6 +73,23 @@ internal sealed class ManagedChildProcess : IAsyncDisposable } } + private async Task ListenForStatusAsync() + { + try + { + while (controlPipe.IsConnected && !Process.HasExited) + { + var status = await controlReader.ReadLineAsync(); + if (status is null) return; + LastStatus = status; + statusReceived?.Invoke(status); + } + } + catch (IOException) when (stopping || Process.HasExited) + { + } + } + internal async Task StopAsync(TimeSpan? timeout = null, CancellationToken cancellationToken = default) { if (stopping || Process.HasExited) return; @@ -77,6 +108,7 @@ internal sealed class ManagedChildProcess : IAsyncDisposable { if (!Process.HasExited) await StopAsync(); controlWriter.Dispose(); + controlReader.Dispose(); controlPipe.Dispose(); Process.Dispose(); } diff --git a/supervisor/Dada.Supervisor/StructuredJsonlLogger.cs b/supervisor/Dada.Supervisor/StructuredJsonlLogger.cs new file mode 100644 index 0000000..3b4adea --- /dev/null +++ b/supervisor/Dada.Supervisor/StructuredJsonlLogger.cs @@ -0,0 +1,158 @@ +using System.Text.Json; + +namespace Dada.Supervisor; + +internal sealed record StructuredLogEvent( + string StatusCategory, + string? CorrelationId = null, + string? ObjectId = null, + long? DurationMs = null, + string? ErrorCategory = null); + +internal static class RedactionPolicy +{ + private static readonly HashSet StatusCategories = + ["starting", "ready", "completed", "failed", "unavailable", "degraded", "blocked", "stopping", "stopped"]; + private static readonly HashSet ErrorCategories = + ["none", "invalid_input", "storage_unavailable", "log_write_failed", "service_unavailable", "timeout", "internal_error"]; + + internal static string StatusCategory(string value) => StatusCategories.Contains(value) ? value : "failed"; + internal static string ErrorCategory(string value) => ErrorCategories.Contains(value) ? value : "internal_error"; + internal static string Code(string value, string fallback) => + value.Length is > 0 and <= 64 && value[0] is >= 'a' and <= 'z' && value.All(character => character is >= 'a' and <= 'z' or >= '0' and <= '9' or '_') + ? value + : fallback; + internal static string? Identifier(string? value) => + value is { Length: > 0 and <= 64 } && value.All(character => char.IsAsciiLetterOrDigit(character) || character is '_' or '-') + ? value + : null; +} + +internal sealed class StructuredJsonlLogger +{ + internal const long SizeLimitBytes = 10 * 1024 * 1024; + internal const int FileLimit = 10; + internal const int RetentionDays = 30; + + private static readonly JsonSerializerOptions JsonOptions = new() + { + PropertyNamingPolicy = JsonNamingPolicy.SnakeCaseLower, + }; + + private readonly string directory; + private readonly string component; + private readonly long sizeLimitBytes; + private readonly int fileLimit; + private readonly Func now; + private readonly Action onWriteFailure; + private static readonly System.Text.Encoding Utf8WithoutBom = new System.Text.UTF8Encoding(false); + private bool initialized; + private long currentSize; + + internal StructuredJsonlLogger( + string directory, + string component, + Action? onWriteFailure = null, + Func? now = null, + long sizeLimitBytes = SizeLimitBytes, + int fileLimit = FileLimit) + { + this.directory = directory; + this.component = component is "api" or "worker" or "supervisor" ? component : "supervisor"; + this.onWriteFailure = onWriteFailure ?? (() => { }); + this.now = now ?? (() => DateTimeOffset.UtcNow); + this.sizeLimitBytes = sizeLimitBytes; + this.fileLimit = fileLimit; + } + + internal void Write(StructuredLogEvent input) + { + try + { + Initialize(); + var entry = new Dictionary + { + ["schema_version"] = "1.0", + ["timestamp"] = now().ToString("O"), + ["component"] = component, + ["status_category"] = RedactionPolicy.StatusCategory(input.StatusCategory), + }; + var correlationId = RedactionPolicy.Identifier(input.CorrelationId); + var objectId = RedactionPolicy.Identifier(input.ObjectId); + if (correlationId is not null) entry["correlation_id"] = correlationId; + if (objectId is not null) entry["object_id"] = objectId; + if (input.DurationMs is >= 0) entry["duration_ms"] = input.DurationMs; + if (input.ErrorCategory is not null) entry["error_category"] = RedactionPolicy.ErrorCategory(input.ErrorCategory); + var line = JsonSerializer.Serialize(entry, JsonOptions) + Environment.NewLine; + var bytes = System.Text.Encoding.UTF8.GetByteCount(line); + if (bytes > sizeLimitBytes) throw new InvalidOperationException("log_entry_too_large"); + if (currentSize > 0 && currentSize + bytes > sizeLimitBytes) Rotate(); + File.AppendAllText(ActivePath(), line, Utf8WithoutBom); + currentSize += bytes; + } + catch (Exception exception) when (exception is not LogWriteException) + { + try { onWriteFailure(); } catch { } + throw new LogWriteException(exception); + } + } + + internal void Maintain() + { + try + { + Directory.CreateDirectory(directory); + var cutoff = now().AddDays(-RetentionDays); + foreach (var file in Files()) + { + if (file.Index > fileLimit - 1 || file.Info.LastWriteTimeUtc < cutoff.UtcDateTime) file.Info.Delete(); + } + foreach (var file in Files().Skip(fileLimit)) file.Info.Delete(); + currentSize = File.Exists(ActivePath()) ? new FileInfo(ActivePath()).Length : 0; + initialized = true; + } + catch (Exception exception) when (exception is not LogWriteException) + { + try { onWriteFailure(); } catch { } + throw new LogWriteException(exception); + } + } + + private void Initialize() + { + if (!initialized) Maintain(); + } + + private void Rotate() + { + File.Delete(Path.Combine(directory, $"{component}.{fileLimit - 1}.jsonl")); + for (var index = fileLimit - 2; index >= 1; index--) + { + var source = Path.Combine(directory, $"{component}.{index}.jsonl"); + if (File.Exists(source)) File.Move(source, Path.Combine(directory, $"{component}.{index + 1}.jsonl")); + } + if (File.Exists(ActivePath())) File.Move(ActivePath(), Path.Combine(directory, $"{component}.1.jsonl")); + currentSize = 0; + } + + private IReadOnlyList<(int Index, FileInfo Info)> Files() + { + if (!Directory.Exists(directory)) return []; + var prefix = component + "."; + return Directory.EnumerateFiles(directory, $"{component}*.jsonl") + .Select(path => + { + var name = Path.GetFileName(path); + var index = name == $"{component}.jsonl" + ? 0 + : int.TryParse(name[prefix.Length..^".jsonl".Length], out var parsed) ? parsed : int.MaxValue; + return (Index: index, Info: new FileInfo(path)); + }) + .OrderBy(file => file.Index) + .ToArray(); + } + + private string ActivePath() => Path.Combine(directory, $"{component}.jsonl"); +} + +internal sealed class LogWriteException(Exception innerException) : IOException("log_write_failed", innerException); diff --git a/supervisor/Dada.Supervisor/SupervisorForm.cs b/supervisor/Dada.Supervisor/SupervisorForm.cs index b552212..bac43fd 100644 --- a/supervisor/Dada.Supervisor/SupervisorForm.cs +++ b/supervisor/Dada.Supervisor/SupervisorForm.cs @@ -268,6 +268,8 @@ internal sealed class SupervisorForm : Form internal sealed class DiagnosticsForm : Form { + internal string CopyPayload { get; } + internal DiagnosticsForm(SupervisorState state) { BackColor = Color.White; @@ -281,11 +283,22 @@ internal sealed class DiagnosticsForm : Form checks.Columns.Add("组件", 170); checks.Columns.Add("状态", 110); checks.Columns.Add("检查结果", 390); + var report = DiagnosticReport.Create( + state, + [ + state == SupervisorState.StorageUnavailable + ? DiagnosticCheck.Fail("log_writable", "log_write_failed") + : DiagnosticCheck.Pass("log_writable", "log_ready"), + DiagnosticCheck.Pass("fixed_port", "loopback_only"), + ], + new DiagnosticStorageSummary(0, 0, 0), + []); + CopyPayload = System.Text.Json.JsonSerializer.Serialize(report, new System.Text.Json.JsonSerializerOptions { WriteIndented = true }); checks.Items.Add(new ListViewItem(["Supervisor", "正常", StateLabel(state)])); checks.Items.Add(new ListViewItem(["固定端口", "43121", "仅绑定 127.0.0.1"])); checks.Items.Add(new ListViewItem(["凭据", "已脱敏", "仅显示是否已配置"])); var copy = new Button { Location = new Point(592, 382), Size = new Size(138, 36), Text = "复制脱敏结果" }; - copy.Click += (_, _) => Clipboard.SetText("Dada diagnostics: redacted"); + copy.Click += (_, _) => Clipboard.SetText(CopyPayload); Controls.AddRange([title, detail, checks, copy]); } diff --git a/supervisor/Dada.Supervisor/SupervisorRuntime.cs b/supervisor/Dada.Supervisor/SupervisorRuntime.cs index 17f35c2..dae1956 100644 --- a/supervisor/Dada.Supervisor/SupervisorRuntime.cs +++ b/supervisor/Dada.Supervisor/SupervisorRuntime.cs @@ -7,6 +7,7 @@ internal sealed class SupervisorRuntime : IAsyncDisposable private readonly ICredentialStore credentials; private ManagedComponentSupervisor? api; private ManagedComponentSupervisor? worker; + private StructuredJsonlLogger? logger; internal SupervisorRuntime(ICredentialStore credentials) { @@ -24,6 +25,15 @@ internal sealed class SupervisorRuntime : IAsyncDisposable { return SupervisorState.StartupFailed; } + try + { + logger = new StructuredJsonlLogger(Path.Combine(configuration.LocalDataRoot, "logs", "supervisor"), "supervisor"); + logger.Write(new StructuredLogEvent("starting", ErrorCategory: "none")); + } + catch (LogWriteException) + { + return SupervisorState.StorageUnavailable; + } if (CredentialCatalog.RequiredFor(ChildRole.Api).Concat(CredentialCatalog.RequiredFor(ChildRole.Worker)).Any(target => !credentials.IsConfigured(target))) { return SupervisorState.StartupFailed; @@ -35,22 +45,50 @@ internal sealed class SupervisorRuntime : IAsyncDisposable if (!File.Exists(node) || !File.Exists(apiEntry) || !File.Exists(workerEntry)) return SupervisorState.StartupFailed; api = CreateComponent(node, apiEntry, ChildRole.Api, SupervisorState.ApiDegraded); - api.StateChanged += state => StateChanged?.Invoke(state); await api.StartAsync(cancellationToken); worker = CreateComponent(node, workerEntry, ChildRole.Worker, SupervisorState.WorkerDegraded); - worker.StateChanged += state => StateChanged?.Invoke(state); await worker.StartAsync(cancellationToken); - return SupervisorState.Ready; + return TryLog(new StructuredLogEvent("ready", ErrorCategory: "none")) ? SupervisorState.Ready : SupervisorState.StorageUnavailable; } - private ManagedComponentSupervisor CreateComponent(string node, string entry, ChildRole role, SupervisorState degradedState) => - new(async cancellationToken => + private ManagedComponentSupervisor CreateComponent(string node, string entry, ChildRole role, SupervisorState degradedState) + { + var component = new ManagedComponentSupervisor(async cancellationToken => { var startInfo = new ProcessStartInfo(node); startInfo.ArgumentList.Add(entry); - return await ManagedChildProcess.StartAsync(startInfo, role, credentials, cancellationToken); + var child = await ManagedChildProcess.StartAsync(startInfo, role, credentials, cancellationToken); + child.StatusReceived += status => + { + if (status == "storage_unavailable") + { + TryLog(new StructuredLogEvent("unavailable", ErrorCategory: "log_write_failed")); + StateChanged?.Invoke(SupervisorState.StorageUnavailable); + } + }; + return child; }, degradedState); + component.StateChanged += state => + { + if (TryLog(new StructuredLogEvent("degraded", ErrorCategory: "service_unavailable"))) StateChanged?.Invoke(state); + }; + return component; + } + + private bool TryLog(StructuredLogEvent entry) + { + try + { + logger?.Write(entry); + return true; + } + catch (LogWriteException) + { + StateChanged?.Invoke(SupervisorState.StorageUnavailable); + return false; + } + } public async ValueTask DisposeAsync() { diff --git a/tests/integration/wp0-08-structured-log.test.ts b/tests/integration/wp0-08-structured-log.test.ts new file mode 100644 index 0000000..ae400af --- /dev/null +++ b/tests/integration/wp0-08-structured-log.test.ts @@ -0,0 +1,137 @@ +import { randomUUID } from "node:crypto"; +import { mkdirSync, readFileSync, readdirSync, statSync, utimesSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; + +import { afterEach, describe, expect, it } from "vitest"; + +import { ManagedStorage } from "../../apps/api/src/managed-storage.js"; +import { + LOG_FILE_LIMIT, + LOG_RETENTION_DAYS, + LOG_SIZE_LIMIT_BYTES, + StructuredJsonlLogger, + sanitizeDiagnosticRecord, +} from "../../apps/api/src/structured-log.js"; + +const temporaryRoots: string[] = []; + +function fixtureRoot() { + const root = join(tmpdir(), `dada-wp0-08-${randomUUID()}`); + mkdirSync(root, { recursive: true }); + temporaryRoots.push(root); + return root; +} + +function evidence(name: string, value: unknown) { + const directory = process.env.DADA_EVIDENCE_DIR_LOG; + if (!directory) return; + mkdirSync(directory, { recursive: true }); + writeFileSync(join(directory, name), `${JSON.stringify(value, null, 2)}\n`); +} + +afterEach(async () => { + const { rm } = await import("node:fs/promises"); + await Promise.all(temporaryRoots.splice(0).map((root) => rm(root, { force: true, recursive: true }))); +}); + +describe("TDD-WP0-LOG-001 rotation and redaction", () => { + it("enforces 10 MiB, 10 files, 30 days and emits only allowlisted JSONL fields", () => { + const root = fixtureRoot(); + const logDirectory = join(root, "logs", "api"); + mkdirSync(logDirectory, { recursive: true }); + const activePath = join(logDirectory, "api.jsonl"); + const seedLine = `${JSON.stringify({ + component: "api", + correlation_id: "corr_fixture", + schema_version: "1.0", + status_category: "ready", + timestamp: "2026-01-01T00:00:00.000Z", + })}\n`; + const repeats = Math.floor(LOG_SIZE_LIMIT_BYTES / Buffer.byteLength(seedLine)); + writeFileSync(activePath, seedLine.repeat(repeats)); + const bytesBeforeBoundaryWrite = statSync(activePath).size; + + const marker = `trap-${randomUUID()}`; + const logger = new StructuredJsonlLogger({ component: "api", directory: logDirectory }); + logger.write({ + absolute_path: join(root, marker), + api_key: `${marker}-key`, + correlation_id: "corr_boundary", + duration_ms: 12, + email: `${marker}@example.invalid`, + error_category: "none", + image: Buffer.from(marker), + object_id: "obj_boundary", + prompt: `${marker} private prompt`, + provider_error: `${marker} upstream body`, + status_category: "completed", + }); + expect(statSync(join(logDirectory, "api.1.jsonl")).size).toBeLessThanOrEqual(LOG_SIZE_LIMIT_BYTES); + expect(statSync(activePath).size).toBeLessThanOrEqual(LOG_SIZE_LIMIT_BYTES); + + for (let index = 2; index <= 14; index += 1) writeFileSync(join(logDirectory, `api.${index}.jsonl`), seedLine); + const stalePath = join(logDirectory, "api.4.jsonl"); + const stale = new Date(Date.now() - (LOG_RETENTION_DAYS + 1) * 24 * 60 * 60 * 1000); + utimesSync(stalePath, stale, stale); + new StructuredJsonlLogger({ component: "api", directory: logDirectory }).maintain(); + + const files = readdirSync(logDirectory).sort(); + const scanned = files.map((name) => readFileSync(join(logDirectory, name), "utf8")).join(""); + expect(files.length).toBeLessThanOrEqual(LOG_FILE_LIMIT); + expect(files).not.toContain("api.4.jsonl"); + expect(scanned).not.toContain(marker); + const activeEntry = JSON.parse(readFileSync(activePath, "utf8").trim()) as Record; + expect(Object.keys(activeEntry).sort()).toEqual([ + "component", "correlation_id", "duration_ms", "error_category", "object_id", "schema_version", "status_category", "timestamp", + ]); + + const diagnostic = sanitizeDiagnosticRecord({ + app_version: "0.0.0", + checked_at: "2026-01-01T00:00:00.000Z", + component: "api", + message_key: "log_write_failed", + private_content: marker, + result: "fail", + stable_check_code: "log_writable", + user_path: join(root, marker), + }); + expect(JSON.stringify(diagnostic)).not.toContain(marker); + evidence("log-manifest.json", { + bytes_before_boundary_write: bytesBeforeBoundaryWrite, + file_limit: LOG_FILE_LIMIT, + files: files.map((name) => ({ name, size: statSync(join(logDirectory, name)).size })), + retention_days: LOG_RETENTION_DAYS, + size_limit_bytes: LOG_SIZE_LIMIT_BYTES, + status: "passed", + total_generated_bytes: bytesBeforeBoundaryWrite + statSync(activePath).size, + }); + evidence("redaction.json", { diagnostic, marker_absent: true, status: "passed" }); + }); + + it("marks storage unavailable when the log directory cannot be written", () => { + const root = fixtureRoot(); + const storage = new ManagedStorage({ dataRoot: root, databasePath: join(root, "db", "dada.sqlite3") }); + const blockedDirectory = join(root, "logs-blocked"); + writeFileSync(blockedDirectory, "not-a-directory"); + const logger = new StructuredJsonlLogger({ + component: "api", + directory: blockedDirectory, + onWriteFailure: () => storage.setLogAvailability(false), + }); + expect(() => logger.write({ correlation_id: "corr_log_failure", error_category: "log_write_failed", status_category: "failed" })).toThrow("log_write_failed"); + expect(storage.getState().storage_status).toBe("unavailable"); + expect(storage.inspectAction("ai_call")).toBe("reject_unavailable"); + evidence("response.json", { + ai_call: storage.inspectAction("ai_call"), + business_objects_changed: 0, + storage_status: storage.getState().storage_status, + status: "passed", + }); + storage.setLogAvailability(true); + expect(storage.getState().storage_status).toBe("unavailable"); + storage.applyControlledMeasurement(0); + expect(storage.getState().storage_status).toBe("active"); + storage.close(); + }); +}); diff --git a/tests/worker/wp0-08-log-availability.test.ts b/tests/worker/wp0-08-log-availability.test.ts new file mode 100644 index 0000000..ea51e67 --- /dev/null +++ b/tests/worker/wp0-08-log-availability.test.ts @@ -0,0 +1,52 @@ +import { randomUUID } from "node:crypto"; +import { mkdirSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; + +import { expect, it, vi } from "vitest"; + +import { ManagedStorage } from "../../apps/api/src/managed-storage.js"; +import { WorkerAiCallGate } from "../../apps/worker/src/ai-call-gate.js"; +import { WorkerStorageStatus } from "../../apps/worker/src/storage-status.js"; +import { StructuredJsonlLogger } from "../../apps/worker/src/structured-log.js"; + +it("TDD-WP0-LOG-001 blocks AI before an upstream call when logging becomes unavailable", async () => { + const root = join(tmpdir(), `dada-wp0-08-worker-${randomUUID()}`); + mkdirSync(root, { recursive: true }); + const blockedDirectory = join(root, "blocked-log-directory"); + writeFileSync(blockedDirectory, "not-a-directory"); + const databasePath = join(root, "db", "dada.sqlite3"); + const managedStorage = new ManagedStorage({ dataRoot: root, databasePath }); + const workerStorage = new WorkerStorageStatus(databasePath); + let storageStatus: "active" | "unavailable" = "active"; + const logger = new StructuredJsonlLogger({ + component: "worker", + directory: blockedDirectory, + onWriteFailure: () => { + storageStatus = "unavailable"; + workerStorage.markLogUnavailable(); + }, + }); + const externalCall = vi.fn(async () => ({ provider_request_id: "must-not-exist" })); + const gate = new WorkerAiCallGate({ getStorageStatus: () => storageStatus, logger }); + + await expect(gate.execute("corr_worker_log_failure", "job_internal", externalCall)).resolves.toEqual({ + reason: "storage_unavailable", + status: "blocked", + }); + expect(storageStatus).toBe("unavailable"); + expect(workerStorage.getStatus()).toBe("unavailable"); + expect(managedStorage.getState().storage_status).toBe("unavailable"); + expect(externalCall).not.toHaveBeenCalled(); + + const directory = process.env.DADA_EVIDENCE_DIR_LOG; + if (directory) { + mkdirSync(directory, { recursive: true }); + writeFileSync(join(directory, "external-calls.json"), `${JSON.stringify({ calls: 0, reason: "storage_unavailable", status: "passed" }, null, 2)}\n`); + } + + const { rm } = await import("node:fs/promises"); + workerStorage.close(); + managedStorage.close(); + await rm(root, { force: true, recursive: true }); +});