Recognise Apple crash frames in the error fingerprint

apps/cli/src/server/archives/journal.ts

@@ -12,6 +12,8 @@ import {
 	validateArchiveId,
 	validateRangeDate,
 } from "./paths"
+import { Schema } from "effect"
+import { NonEmptyString, NonNegativeSafeInt, Sha256Lower } from "./schemas"
 import { parseArchiveActivePointer } from "./manifest"
 import { archiveSignal } from "./signals"
 
@@ -177,6 +179,19 @@ const intentPath = (archiveDir: string, operationId: string): string =>
 const isRecord = (value: unknown): value is Record<string, unknown> =>
 	typeof value === "object" && value !== null
 
+const decodeCountValue = Schema.decodeUnknownSync(NonNegativeSafeInt)
+
+/** Same rule as the GC targets' byte counts, with the field named in the error. */
+const decodeCount = (value: unknown, field: string): number => {
+	try {
+		return decodeCountValue(value)
+	} catch {
+		throw new Error(`archive operation gc intent has invalid ${field}: ${String(value)}`)
+	}
+}
+
+const decodeSha256 = Schema.decodeUnknownSync(Sha256Lower)
+
 const requiredString = (value: unknown, field: string): string => {
 	if (typeof value !== "string" || value.length === 0)
 		throw new Error(`journal field ${field} missing or not a string`)
@@ -289,9 +304,10 @@ const parseCreateIntent = (
 		throw new Error(`archive operation pin purpose does not match generation: ${pinPurpose}`)
 	}
 	const manifestSha256Raw = raw.manifestSha256
-	const manifestSha256 =
-		manifestSha256Raw === null ? null : requiredString(manifestSha256Raw, "manifestSha256").toLowerCase()
-	if (manifestSha256 !== null && !/^[0-9a-f]{64}$/.test(manifestSha256)) {
+	let manifestSha256: string | null
+	try {
+		manifestSha256 = manifestSha256Raw === null ? null : decodeSha256(manifestSha256Raw)
+	} catch {
 		throw new Error("invalid archive operation manifestSha256")
 	}
 	if (phaseRequiresManifest(phase) !== (manifestSha256 !== null)) {
@@ -337,20 +353,8 @@ const parseGcIntent = (
 	createdAt: string,
 	updatedAt: string,
 ): GcOperationIntent => {
-	const keepRaw = raw.keep
-	if (typeof keepRaw !== "number" || !Number.isSafeInteger(keepRaw) || keepRaw < 0) {
-		throw new Error(`archive operation gc intent has invalid keep: ${String(keepRaw)}`)
-	}
-	const completedTargetsRaw = raw.completedTargets
-	if (
-		typeof completedTargetsRaw !== "number" ||
-		!Number.isSafeInteger(completedTargetsRaw) ||
-		completedTargetsRaw < 0
-	) {
-		throw new Error(
-			`archive operation gc intent has invalid completedTargets: ${String(completedTargetsRaw)}`,
-		)
-	}
+	const keepRaw = decodeCount(raw.keep, "keep")
+	const completedTargetsRaw = decodeCount(raw.completedTargets, "completedTargets")
 	const targetsRaw = raw.targets
 	if (!Array.isArray(targetsRaw)) {
 		throw new Error("archive operation gc intent targets is not an array")
@@ -402,50 +406,54 @@ const parseGcIntent = (
 	}
 }
 
+/**
+ * One GC target exactly as the journal recorded it.
+ *
+ * Only the recorded fields are decoded here. The identities are re-validated
+ * through the same
+ * canonicalizing validators the writer used, because a hand-edited journal must
+ * not be able to point collection at an arbitrary path.
+ */
+const RecordedGcTarget = Schema.Struct({
+	signal: NonEmptyString,
+	rangeStart: NonEmptyString,
+	generationId: NonEmptyString,
+	createdAt: NonEmptyString,
+	manifestSha256: Sha256Lower,
+	bytes: NonNegativeSafeInt,
+	recordedActiveGenerationId: NonEmptyString,
+	shards: Schema.Array(
+		Schema.Struct({
+			name: NonEmptyString,
+			bytes: NonNegativeSafeInt,
+			sha256: Sha256Lower,
+		}),
+	),
+})
+
+const decodeRecordedGcTarget = Schema.decodeUnknownSync(RecordedGcTarget)
+
 const parseGcTarget = (raw: unknown, index: number): GcTarget => {
-	if (!isRecord(raw)) throw new Error(`archive gc target ${index} is not a record`)
-	const signal = archiveSignal(requiredString(raw.signal, "signal")).name
-	const rangeStart = validateRangeDate(requiredString(raw.rangeStart, "rangeStart"))
-	const generationId = validateArchiveId(requiredString(raw.generationId, "generationId"), "generation")
-	const createdAt = requiredString(raw.createdAt, "createdAt")
-	const manifestSha256 = requiredString(raw.manifestSha256, "manifestSha256").toLowerCase()
-	if (!/^[0-9a-f]{64}$/.test(manifestSha256)) {
-		throw new Error(`archive gc target ${index} has invalid manifestSha256`)
-	}
-	const bytesRaw = raw.bytes
-	if (typeof bytesRaw !== "number" || !Number.isSafeInteger(bytesRaw) || bytesRaw < 0) {
-		throw new Error(`archive gc target ${index} has invalid bytes: ${String(bytesRaw)}`)
+	let recorded: typeof RecordedGcTarget.Type
+	try {
+		recorded = decodeRecordedGcTarget(raw)
+	} catch (error) {
+		throw new Error(
+			`archive gc target ${index} is malformed: ${error instanceof Error ? error.message : String(error)}`,
+		)
 	}
-	const recordedActiveGenerationId = validateArchiveId(
-		requiredString(raw.recordedActiveGenerationId, "recordedActiveGenerationId"),
-		"active generation",
-	)
-	const shardsRaw = raw.shards
-	if (!Array.isArray(shardsRaw)) {
-		throw new Error(`archive gc target ${index} shards is not an array`)
-	}
-	const shards = shardsRaw.map((s, j) => {
-		if (!isRecord(s)) throw new Error(`archive gc target ${index} shard ${j} is not a record`)
-		const name = requiredString(s.name, "name")
-		const bytes = s.bytes
-		if (typeof bytes !== "number" || !Number.isSafeInteger(bytes) || bytes < 0) {
-			throw new Error(`archive gc target ${index} shard ${j} invalid bytes`)
-		}
-		const sha256 = requiredString(s.sha256, "sha256").toLowerCase()
-		if (!/^[0-9a-f]{64}$/.test(sha256)) {
-			throw new Error(`archive gc target ${index} shard ${j} invalid sha256`)
-		}
-		return { name, bytes, sha256 }
-	})
 	return {
-		signal,
-		rangeStart,
-		generationId,
-		createdAt,
-		manifestSha256,
-		bytes: bytesRaw,
-		shards,
-		recordedActiveGenerationId,
+		signal: archiveSignal(recorded.signal).name,
+		rangeStart: validateRangeDate(recorded.rangeStart),
+		generationId: validateArchiveId(recorded.generationId, "generation"),
+		createdAt: recorded.createdAt,
+		manifestSha256: recorded.manifestSha256,
+		bytes: recorded.bytes,
+		shards: recorded.shards,
+		recordedActiveGenerationId: validateArchiveId(
+			recorded.recordedActiveGenerationId,
+			"active generation",
+		),
 	}
 }
 

apps/cli/src/server/archives/manifest.ts

@@ -1,11 +1,13 @@
 import { readFileSync } from "node:fs"
 import { join } from "node:path"
+import { Schema } from "effect"
 import {
 	type ArchiveTuningRecord,
 	type TuningConfigIdentity,
 	LEGACY_TUNING_CONFIG_FORMAT_VERSION,
 	TUNING_CONFIG_FORMAT_VERSION,
 } from "./config"
+import { NonNegativeSafeInt, Sha256Lower } from "./schemas"
 import { KNOWN_COMPLEX_DIGEST_ALGORITHMS } from "./export"
 import {
 	assertNoSymlinkSync,
@@ -150,12 +152,14 @@ const requiredString = (record: Record<string, unknown>, key: string): string =>
 	return value
 }
 
+const decodeCount = Schema.decodeUnknownSync(NonNegativeSafeInt)
+
 const requiredCount = (record: Record<string, unknown>, key: string): number => {
-	const value = record[key]
-	if (typeof value !== "number" || !Number.isSafeInteger(value) || value < 0) {
+	try {
+		return decodeCount(record[key])
+	} catch {
 		throw new Error(`invalid archive manifest field: ${key} (must be a safe non-negative integer)`)
 	}
-	return value
 }
 
 const requiredPositiveInteger = (record: Record<string, unknown>, key: string): number => {
@@ -166,7 +170,9 @@ const requiredPositiveInteger = (record: Record<string, unknown>, key: string):
 	return value
 }
 
-const SHA256_HEX = /^[0-9a-f]{64}$/
+/** The same 64-hex rule the journal enforces, stated once in ./schemas. */
+const isSha256Hex = (value: string): boolean =>
+	Schema.decodeUnknownResult(Sha256Lower)(value)._tag === "Success"
 
 /** A safe logical config name (no path separators, no traversal). */
 const SAFE_CONFIG_NAME = /^[A-Za-z0-9._-]+$/
@@ -208,7 +214,7 @@ const parseTuningConfig = (value: unknown): TuningConfigIdentity | null => {
 		throw new Error(`invalid archive manifest tuningConfig.configName (unsafe name): ${configName}`)
 	}
 	const sha256 = requiredString(value, "sha256")
-	if (!SHA256_HEX.test(sha256)) {
+	if (!isSha256Hex(sha256)) {
 		throw new Error(`invalid archive manifest tuningConfig.sha256 (must be 64 hex chars): ${sha256}`)
 	}
 	return { formatVersion, configName, sha256 }
@@ -297,7 +303,7 @@ const parseShardRecord = (
 		return c
 	})
 	const sha256 = requiredString(value, "sha256")
-	if (!SHA256_HEX.test(sha256))
+	if (!isSha256Hex(sha256))
 		throw new Error(`invalid archive shard sha256 (must be 64 hex chars): ${sha256}`)
 	const rowCount = requiredCount(value, "rowCount")
 	const minNano = requiredNanoDecimal(value, "minEventTimeUnixNano")

apps/cli/src/server/archives/schemas.ts

@@ -0,0 +1,36 @@
+// Shape primitives shared by the archive journal and manifest readers.
+//
+// These are the rules that were genuinely copy-pasted: a non-negative safe
+// integer was re-checked at four sites in the GC intent alone, and the sha256
+// pattern at six across the two files. The rest of the archive parsers are
+// domain invariants — path containment, phase-vs-kind compatibility, cursor
+// consistency — each with its own tested message, and those stay where they
+// are rather than being flattened into a struct declaration.
+import { Schema, SchemaGetter } from "effect"
+
+/** Byte counts and cursors: never negative, never beyond exact integer range. */
+export const NonNegativeSafeInt = Schema.Number.check(
+	Schema.makeFilter((value: number) =>
+		Number.isSafeInteger(value) && value >= 0 ? undefined : "must be a safe non-negative integer",
+	),
+)
+
+/**
+ * A sha256 digest, normalized to lower case before it is compared.
+ *
+ * Case normalization is part of the rule, not a courtesy: these digests gate
+ * whether a generation may be deleted, and an upper-case digest that failed to
+ * compare equal would look like corruption of data that is in fact intact.
+ */
+export const Sha256Lower = Schema.String.pipe(
+	Schema.decodeTo(Schema.String, {
+		decode: SchemaGetter.transform((value: string) => value.toLowerCase()),
+		encode: SchemaGetter.passthrough(),
+	}),
+).check(
+	Schema.makeFilter((value: string) =>
+		/^[0-9a-f]{64}$/.test(value) ? undefined : "must be 64 hex characters",
+	),
+)
+
+export const NonEmptyString = Schema.String.check(Schema.isMinLength(1))

apps/cli/src/server/chdb-rows.ts

@@ -0,0 +1,61 @@
+// Decoding rows out of chDB's JSONEachRow output.
+//
+// Nine copies of a `split("\n") … JSON.parse(line) as A` helper used to live
+// across the migration coordinator, the physical-schema inspector, and every
+// versioned edge — each one ending in an unchecked cast at the point where an
+// external process hands us bytes. The shape is declared once here and the
+// rows are decoded, not asserted.
+import { Schema } from "effect"
+
+/**
+ * Decode chDB JSONEachRow output.
+ *
+ * A row that does not match the schema is a query returning something other
+ * than what the caller asked for, which is a bug in the SQL rather than data to
+ * be tolerated — so this throws instead of skipping the row.
+ */
+export const decodeJsonEachRow = <S extends Schema.Codec<unknown, unknown, never, never>>(
+	rowSchema: S,
+): ((value: string) => Array<S["Type"]>) => {
+	const decodeRow = Schema.decodeUnknownSync(rowSchema)
+	return (value) =>
+		value
+			.split("\n")
+			.map((line) => line.trim())
+			.filter((line) => line.length > 0)
+			.map((line) => decodeRow(JSON.parse(line)))
+}
+
+/**
+ * `sum(rows)` per table, as returned by the raw-telemetry inventory query.
+ *
+ * The count is `toString()`-wrapped in SQL because it is a ClickHouse UInt64:
+ * above 2^53 a JS number silently loses the low bits, and these values are
+ * compared for exact equality when a migration verifies that no row was lost.
+ */
+export const TableRowCountSchema = Schema.Struct({
+	table: Schema.String,
+	rowCount: Schema.String.check(Schema.isPattern(/^\d+$/)),
+})
+
+export const decodeTableRowCounts = decodeJsonEachRow(TableRowCountSchema)
+
+/** A single `toString(count())`-style scalar, for the same UInt64 reason. */
+export const RowCountSchema = Schema.Struct({
+	rowCount: Schema.String.check(Schema.isPattern(/^\d+$/)),
+})
+
+export const decodeRowCounts = decodeJsonEachRow(RowCountSchema)
+
+/**
+ * Rows whose columns are not known ahead of time.
+ *
+ * The v0 -> v1 raw replay copies whatever columns the source table happens to
+ * have, so there is no field list to declare. Requiring each line to be a JSON
+ * object is the only claim that can honestly be made about it, and it is still
+ * a claim worth making: a bare scalar or array here means the query returned
+ * something other than rows.
+ */
+const OpaqueRowSchema = Schema.Record(Schema.String, Schema.Unknown)
+
+export const decodeJsonObjectRows = decodeJsonEachRow(OpaqueRowSchema)

apps/cli/src/server/chdb.ts

@@ -104,32 +104,34 @@ export const RAW_TELEMETRY_TTL_COLUMNS = [
 export const MINIMUM_RAW_TELEMETRY_RETENTION_DAYS = 90
 export const MAXIMUM_RAW_TELEMETRY_RETENTION_DAYS = 3_650
 
-interface RawTelemetryRetentionConfig {
-	readonly formatVersion: 1
-	readonly minimumDays: number
-}
+/**
+ * The retention floor an operator has pinned for this store.
+ *
+ * Unknown fields are rejected rather than ignored: a config carrying a field
+ * this build does not understand was written by a different build, and reading
+ * only the half we recognise would silently apply a policy nobody chose.
+ */
+const RawTelemetryRetentionConfigSchema = Schema.Struct({
+	formatVersion: Schema.Literal(1),
+	minimumDays: Schema.Int.check(
+		Schema.makeFilter((days: number) =>
+			days >= MINIMUM_RAW_TELEMETRY_RETENTION_DAYS && days <= MAXIMUM_RAW_TELEMETRY_RETENTION_DAYS
+				? undefined
+				: `raw telemetry retention minimum must be an integer from ${MINIMUM_RAW_TELEMETRY_RETENTION_DAYS} through ${MAXIMUM_RAW_TELEMETRY_RETENTION_DAYS} days`,
+		),
+	),
+})
+
+type RawTelemetryRetentionConfig = typeof RawTelemetryRetentionConfigSchema.Type
 
 export const rawTelemetryRetentionConfigPath = (dataDir: string): string =>
 	`${resolve(dataDir)}.raw-telemetry-retention.json`
 
-const parseRawTelemetryRetentionDays = (value: unknown): number => {
-	if (typeof value !== "object" || value === null || Array.isArray(value))
-		throw new Error("raw telemetry retention config must be a record")
-	const record = value as Record<string, unknown>
-	if (Object.keys(record).sort().join(",") !== "formatVersion,minimumDays" || record.formatVersion !== 1)
-		throw new Error("unsupported or malformed raw telemetry retention config")
-	const days = record.minimumDays
-	if (
-		typeof days !== "number" ||
-		!Number.isSafeInteger(days) ||
-		days < MINIMUM_RAW_TELEMETRY_RETENTION_DAYS ||
-		days > MAXIMUM_RAW_TELEMETRY_RETENTION_DAYS
-	)
-		throw new Error(
-			`raw telemetry retention minimum must be an integer from ${MINIMUM_RAW_TELEMETRY_RETENTION_DAYS} through ${MAXIMUM_RAW_TELEMETRY_RETENTION_DAYS} days`,
-		)
-	return days
-}
+const decodeRetentionConfig = Schema.decodeUnknownSync(RawTelemetryRetentionConfigSchema, {
+	onExcessProperty: "error",
+})
+
+const parseRawTelemetryRetentionDays = (value: unknown): number => decodeRetentionConfig(value).minimumDays
 
 export const readRawTelemetryRetentionDays = (dataDir: string): number | undefined => {
 	const path = rawTelemetryRetentionConfigPath(dataDir)