Add optional DuckDB debug log database with typed schema

New --debug-db flag writes the redacted debug record stream to a DuckDB database alongside or instead of the NDJSON --debug-file. Frequent type-stable attributes are promoted to typed columns; the complete record remains in attrs JSON. libduckdb is dlopen'ed at runtime (build tag duckdb, Arch optdepends) so the reader runs unchanged without it. v1 databases upgrade in place on append.

internal/logging/duckdb.go

@@ -0,0 +1,381 @@
+// Cthulhu screen reader.
+// Copyright (c) 2026 Storm Dragon
+// Authors: Storm Dragon, Angus Kola
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free
+// Software Foundation, either version 3 of the License, or (at your option)
+// any later version.
+//
+// This program is distributed in the hope that it will be useful, but WITHOUT
+// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+// FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+// more details.
+//
+// You should have received a copy of the GNU General Public License along with
+// this program. If not, see <https://www.gnu.org/licenses/gpl-3.0.html>.
+
+package logging
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"strings"
+	"time"
+)
+
+// ErrDuckDBUnavailable reports that DuckDB debug logging cannot be used in
+// this build or on this system: either the build was compiled without
+// -tags duckdb, or libduckdb could not be loaded at runtime.
+var ErrDuckDBUnavailable = errors.New("duckdb debug logging unavailable (requires a -tags duckdb build and an installed libduckdb)")
+
+// Debug database schema. The fields Handler.format emits for every record
+// become typed columns, and so do the record attributes listed in
+// promotedAttrs: keys that real logs show are frequent, scalar, and
+// type-stable, which makes them honest filter and aggregate dimensions for
+// SQL (real types, column statistics, no JSON extraction). Everything else
+// is preserved verbatim in the attrs JSON column, promoted keys included,
+// because the full key set is open-ended (over 100 keys, nested objects
+// and arrays, occasional type changes), so one column per key is not
+// viable. Promoted columns follow attrs so a v1 database upgrades in
+// place by appending them.
+const (
+	duckDBSchemaVersion   = "2"
+	duckDBSchemaVersionV1 = "1"
+
+	duckDBCreateMeta = `CREATE TABLE IF NOT EXISTS meta (
+		key VARCHAR PRIMARY KEY,
+		value VARCHAR NOT NULL
+	)`
+	duckDBCreateSessions = `CREATE TABLE IF NOT EXISTS sessions (
+		session_id VARCHAR PRIMARY KEY,
+		started_at TIMESTAMPTZ NOT NULL,
+		pid BIGINT NOT NULL
+	)`
+)
+
+// Column order here is the appender order in duckdb_native.go.
+var duckDBCreateLogs = buildLogsDDL()
+
+func buildLogsDDL() string {
+	var ddl strings.Builder
+	ddl.WriteString(`CREATE TABLE IF NOT EXISTS logs (
+		seq BIGINT NOT NULL,
+		session_id VARCHAR NOT NULL,
+		ts TIMESTAMPTZ NOT NULL,
+		level VARCHAR NOT NULL,
+		message VARCHAR NOT NULL,
+		source VARCHAR,
+		function VARCHAR,
+		attrs JSON NOT NULL`)
+	for _, attr := range promotedAttrs {
+		fmt.Fprintf(&ddl, ",\n\t\t\"%s\" %s", attr.key, attr.sqlType())
+	}
+	ddl.WriteString("\n\t)")
+	return ddl.String()
+}
+
+type promotedKind int
+
+const (
+	promotedText promotedKind = iota
+	promotedInt
+	promotedBool
+)
+
+type promotedAttr struct {
+	key  string
+	kind promotedKind
+}
+
+func (attr promotedAttr) sqlType() string {
+	switch attr.kind {
+	case promotedInt:
+		return "BIGINT"
+	case promotedBool:
+		return "BOOLEAN"
+	default:
+		return "VARCHAR"
+	}
+}
+
+// promotedAttrs was derived by profiling real debug sessions: every key
+// here is scalar and single-typed (null aside) at meaningful frequency, or
+// carries errors. Free-text payloads (utterance, speech_text) stay in
+// attrs only; they are content, not dimensions.
+var promotedAttrs = []promotedAttr{
+	{key: "role", kind: promotedText},
+	{key: "type", kind: promotedText},
+	{key: "bus", kind: promotedText},
+	{key: "source_path", kind: promotedText},
+	{key: "application", kind: promotedText},
+	{key: "window_path", kind: promotedText},
+	{key: "native_type", kind: promotedText},
+	{key: "origin_path", kind: promotedText},
+	{key: "window_bus", kind: promotedText},
+	{key: "focus_outside_document", kind: promotedText},
+	{key: "radio_group_name", kind: promotedText},
+	{key: "editable_root", kind: promotedText},
+	{key: "read_only_combo", kind: promotedText},
+	{key: "key", kind: promotedText},
+	{key: "cause", kind: promotedText},
+	{key: "reason", kind: promotedText},
+	{key: "command", kind: promotedText},
+	{key: "object_path", kind: promotedText},
+	{key: "capture_reason", kind: promotedText},
+	{key: "error", kind: promotedText},
+	{key: "component", kind: promotedText},
+	{key: "service", kind: promotedText},
+
+	{key: "window_recovered", kind: promotedBool},
+	{key: "focused", kind: promotedBool},
+	{key: "browser_tab", kind: promotedBool},
+	{key: "selected", kind: promotedBool},
+	{key: "checked", kind: promotedBool},
+	{key: "prompt_message_present", kind: promotedBool},
+	{key: "editable", kind: promotedBool},
+	{key: "autocomplete", kind: promotedBool},
+	{key: "autocomplete_popup", kind: promotedBool},
+	{key: "native_caret_known", kind: promotedBool},
+	{key: "input_repeat", kind: promotedBool},
+	{key: "append", kind: promotedBool},
+	{key: "interrupt_previous", kind: promotedBool},
+	{key: "character", kind: promotedBool},
+	{key: "custom_echo", kind: promotedBool},
+	{key: "moved", kind: promotedBool},
+	{key: "has_offset", kind: promotedBool},
+
+	{key: "snapshot_duration", kind: promotedInt},
+	{key: "event_detail1", kind: promotedInt},
+	{key: "native_caret_offset", kind: promotedInt},
+	{key: "native_line_start", kind: promotedInt},
+	{key: "character_count", kind: promotedInt},
+	{key: "selection_start", kind: promotedInt},
+	{key: "selection_end", kind: promotedInt},
+	{key: "input_timestamp", kind: promotedInt},
+	{key: "recent", kind: promotedInt},
+	{key: "input_keycode", kind: promotedInt},
+	{key: "read_all_index", kind: promotedInt},
+	{key: "capture_duration", kind: promotedInt},
+	{key: "speech_style", kind: promotedInt},
+	{key: "unit", kind: promotedInt},
+	{key: "offset", kind: promotedInt},
+	{key: "elapsed", kind: promotedInt},
+	{key: "duration", kind: promotedInt},
+	{key: "stage_duration", kind: promotedInt},
+	{key: "captured_nodes", kind: promotedInt},
+	{key: "pending_nodes", kind: promotedInt},
+	{key: "text_bytes", kind: promotedInt},
+}
+
+// promotedValue is one promotedAttrs entry decoded from a record. valid is
+// false when the key is absent, JSON null, or holds a different type than
+// the column; the value still survives in attrs in every case.
+type promotedValue struct {
+	text    string
+	integer int64
+	boolean bool
+	valid   bool
+}
+
+func decodePromoted(kind promotedKind, raw json.RawMessage) promotedValue {
+	if string(raw) == "null" {
+		return promotedValue{}
+	}
+	switch kind {
+	case promotedText:
+		var value string
+		if json.Unmarshal(raw, &value) == nil {
+			return promotedValue{text: value, valid: true}
+		}
+	case promotedInt:
+		var value int64
+		if json.Unmarshal(raw, &value) == nil {
+			return promotedValue{integer: value, valid: true}
+		}
+	case promotedBool:
+		var value bool
+		if json.Unmarshal(raw, &value) == nil {
+			return promotedValue{boolean: value, valid: true}
+		}
+	}
+	return promotedValue{}
+}
+
+// parsedRecord is one formatted log line split into typed columns and the
+// remaining attributes as JSON object text.
+type parsedRecord struct {
+	timestamp time.Time
+	level     string
+	message   string
+	source    string
+	function  string
+	attrs     string
+	promoted  []promotedValue
+}
+
+// parseLogRecord splits one NDJSON line produced by Handler.format. The
+// formatter emits message, level, source and function first, then any
+// record attributes, and time last of all. An attribute may reuse one of
+// those names (for example an integer "level" attribute), so the leading
+// fields are owned by their first occurrence and time by its last;
+// superseded duplicates fall through to attrs. ok is false for lines that
+// do not match the formatter's shape, in which case the caller stores the
+// raw line instead of dropping it.
+func parseLogRecord(line []byte) (record parsedRecord, ok bool) {
+	decoder := json.NewDecoder(bytes.NewReader(line))
+	token, err := decoder.Token()
+	if err != nil {
+		return parsedRecord{}, false
+	}
+	if delim, isDelim := token.(json.Delim); !isDelim || delim != '{' {
+		return parsedRecord{}, false
+	}
+	attrs := map[string]json.RawMessage{}
+	seen := map[string]bool{}
+	var timeRaw json.RawMessage
+	for decoder.More() {
+		keyToken, err := decoder.Token()
+		if err != nil {
+			return parsedRecord{}, false
+		}
+		key, isString := keyToken.(string)
+		if !isString {
+			return parsedRecord{}, false
+		}
+		var raw json.RawMessage
+		if err := decoder.Decode(&raw); err != nil {
+			return parsedRecord{}, false
+		}
+		if key == "time" {
+			// The formatter's timestamp comes last; any earlier "time"
+			// attribute is demoted to attrs when the real one arrives.
+			if timeRaw != nil {
+				attrs["time"] = timeRaw
+			}
+			timeRaw = raw
+			continue
+		}
+		if isReservedLogKey(key) && !seen[key] {
+			seen[key] = true
+			if !record.setReserved(key, raw) {
+				return parsedRecord{}, false
+			}
+			continue
+		}
+		attrs[key] = raw
+	}
+	if _, err := decoder.Token(); err != nil { // closing brace
+		return parsedRecord{}, false
+	}
+	if _, err := decoder.Token(); err != io.EOF { // trailing newline only
+		return parsedRecord{}, false
+	}
+	if !seen["message"] || !seen["level"] || timeRaw == nil {
+		return parsedRecord{}, false
+	}
+	var timeText string
+	if err := json.Unmarshal(timeRaw, &timeText); err != nil {
+		return parsedRecord{}, false
+	}
+	parsedTime, err := time.Parse(time.RFC3339Nano, timeText)
+	if err != nil {
+		return parsedRecord{}, false
+	}
+	record.timestamp = parsedTime
+	if len(attrs) == 0 {
+		record.attrs = "{}"
+	} else {
+		encoded, err := json.Marshal(attrs)
+		if err != nil {
+			return parsedRecord{}, false
+		}
+		record.attrs = string(encoded)
+	}
+	record.promoted = make([]promotedValue, len(promotedAttrs))
+	for i, attr := range promotedAttrs {
+		if raw, present := attrs[attr.key]; present {
+			record.promoted[i] = decodePromoted(attr.kind, raw)
+		}
+	}
+	return record, true
+}
+
+func isReservedLogKey(key string) bool {
+	switch key {
+	case "message", "level", "source", "function":
+		return true
+	}
+	return false
+}
+
+func (record *parsedRecord) setReserved(key string, raw json.RawMessage) bool {
+	var text string
+	if err := json.Unmarshal(raw, &text); err != nil {
+		return false
+	}
+	switch key {
+	case "message":
+		record.message = text
+	case "level":
+		record.level = text
+	case "source":
+		record.source = text
+	case "function":
+		record.function = text
+	}
+	return true
+}
+
+// multiWriter fans each formatted record out to every sink, for example the
+// NDJSON file and the DuckDB database at the same time. A failing sink does
+// not starve the others; errors are joined and reported to the diagnostics
+// loop, which records them like any other write error.
+type multiWriter struct {
+	writers []DiagnosticsWriter
+}
+
+func NewMultiWriter(writers ...DiagnosticsWriter) DiagnosticsWriter {
+	live := make([]DiagnosticsWriter, 0, len(writers))
+	for _, writer := range writers {
+		if writer != nil {
+			live = append(live, writer)
+		}
+	}
+	switch len(live) {
+	case 0:
+		return NewPassthroughWriter(nil)
+	case 1:
+		return live[0]
+	}
+	return &multiWriter{writers: live}
+}
+
+func (writer *multiWriter) Write(data []byte) (int, error) {
+	var joined error
+	for _, sink := range writer.writers {
+		if _, err := sink.Write(data); err != nil {
+			joined = errors.Join(joined, err)
+		}
+	}
+	return len(data), joined
+}
+
+func (writer *multiWriter) Cancel() error {
+	var joined error
+	for _, sink := range writer.writers {
+		joined = errors.Join(joined, sink.Cancel())
+	}
+	return joined
+}
+
+func (writer *multiWriter) Close() error {
+	var joined error
+	for _, sink := range writer.writers {
+		joined = errors.Join(joined, sink.Close())
+	}
+	return joined
+}

internal/logging/duckdb_native.go

@@ -0,0 +1,648 @@
+//go:build duckdb && cgo
+
+// Cthulhu screen reader.
+// Copyright (c) 2026 Storm Dragon
+// Authors: Storm Dragon, Angus Kola
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free
+// Software Foundation, either version 3 of the License, or (at your option)
+// any later version.
+//
+// This program is distributed in the hope that it will be useful, but WITHOUT
+// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+// FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+// more details.
+//
+// You should have received a copy of the GNU General Public License along with
+// this program. If not, see <https://www.gnu.org/licenses/gpl-3.0.html>.
+
+package logging
+
+/*
+#cgo LDFLAGS: -ldl
+#include <duckdb.h>
+#include <dlfcn.h>
+#include <stdint.h>
+#include <stdlib.h>
+
+// libduckdb is an optional runtime dependency (an Arch optdepend): the
+// reader must start and run on systems without it, so the library is
+// dlopen'ed only when --debug-db is actually used instead of being linked.
+// The header is still needed at build time; the pointers below take their
+// types from it, so calls through them are checked against the installed
+// header exactly like direct calls would be.
+
+#define CTH_DUCK_PTR(name) static __typeof__(&name) cth_p_##name
+CTH_DUCK_PTR(duckdb_create_config);
+CTH_DUCK_PTR(duckdb_set_config);
+CTH_DUCK_PTR(duckdb_destroy_config);
+CTH_DUCK_PTR(duckdb_open_ext);
+CTH_DUCK_PTR(duckdb_free);
+CTH_DUCK_PTR(duckdb_connect);
+CTH_DUCK_PTR(duckdb_query);
+CTH_DUCK_PTR(duckdb_result_error);
+CTH_DUCK_PTR(duckdb_destroy_result);
+CTH_DUCK_PTR(duckdb_row_count);
+CTH_DUCK_PTR(duckdb_value_varchar);
+CTH_DUCK_PTR(duckdb_appender_create);
+CTH_DUCK_PTR(duckdb_appender_error);
+CTH_DUCK_PTR(duckdb_appender_begin_row);
+CTH_DUCK_PTR(duckdb_append_int64);
+CTH_DUCK_PTR(duckdb_append_varchar_length);
+CTH_DUCK_PTR(duckdb_create_timestamp_tz);
+CTH_DUCK_PTR(duckdb_append_value);
+CTH_DUCK_PTR(duckdb_destroy_value);
+CTH_DUCK_PTR(duckdb_append_null);
+CTH_DUCK_PTR(duckdb_append_bool);
+CTH_DUCK_PTR(duckdb_appender_end_row);
+CTH_DUCK_PTR(duckdb_appender_flush);
+CTH_DUCK_PTR(duckdb_appender_close);
+CTH_DUCK_PTR(duckdb_appender_destroy);
+CTH_DUCK_PTR(duckdb_disconnect);
+CTH_DUCK_PTR(duckdb_close);
+
+// cth_duckdb_load resolves every symbol the sink uses from the library at
+// path. It returns NULL on success or a diagnostic string (dlerror output or
+// a string literal; neither must be freed) on failure.
+static const char *cth_duckdb_load(const char *path) {
+	void *handle = dlopen(path, RTLD_NOW | RTLD_LOCAL);
+	if (!handle) return dlerror();
+#define CTH_DUCK_SYM(name) do { \
+		*(void **)(&cth_p_##name) = dlsym(handle, #name); \
+		if (!cth_p_##name) return "missing symbol " #name; \
+	} while (0)
+	CTH_DUCK_SYM(duckdb_create_config);
+	CTH_DUCK_SYM(duckdb_set_config);
+	CTH_DUCK_SYM(duckdb_destroy_config);
+	CTH_DUCK_SYM(duckdb_open_ext);
+	CTH_DUCK_SYM(duckdb_free);
+	CTH_DUCK_SYM(duckdb_connect);
+	CTH_DUCK_SYM(duckdb_query);
+	CTH_DUCK_SYM(duckdb_result_error);
+	CTH_DUCK_SYM(duckdb_destroy_result);
+	CTH_DUCK_SYM(duckdb_row_count);
+	CTH_DUCK_SYM(duckdb_value_varchar);
+	CTH_DUCK_SYM(duckdb_appender_create);
+	CTH_DUCK_SYM(duckdb_appender_error);
+	CTH_DUCK_SYM(duckdb_appender_begin_row);
+	CTH_DUCK_SYM(duckdb_append_int64);
+	CTH_DUCK_SYM(duckdb_append_varchar_length);
+	CTH_DUCK_SYM(duckdb_create_timestamp_tz);
+	CTH_DUCK_SYM(duckdb_append_value);
+	CTH_DUCK_SYM(duckdb_destroy_value);
+	CTH_DUCK_SYM(duckdb_append_null);
+	CTH_DUCK_SYM(duckdb_append_bool);
+	CTH_DUCK_SYM(duckdb_appender_end_row);
+	CTH_DUCK_SYM(duckdb_appender_flush);
+	CTH_DUCK_SYM(duckdb_appender_close);
+	CTH_DUCK_SYM(duckdb_appender_destroy);
+	CTH_DUCK_SYM(duckdb_disconnect);
+	CTH_DUCK_SYM(duckdb_close);
+#undef CTH_DUCK_SYM
+	return NULL;
+}
+
+static duckdb_state cth_duckdb_create_config(duckdb_config *out) {
+	return cth_p_duckdb_create_config(out);
+}
+static duckdb_state cth_duckdb_set_config(duckdb_config config, const char *name, const char *option) {
+	return cth_p_duckdb_set_config(config, name, option);
+}
+static void cth_duckdb_destroy_config(duckdb_config *config) {
+	cth_p_duckdb_destroy_config(config);
+}
+static duckdb_state cth_duckdb_open_ext(const char *path, duckdb_database *out, duckdb_config config, char **err) {
+	return cth_p_duckdb_open_ext(path, out, config, err);
+}
+static void cth_duckdb_free(void *pointer) {
+	cth_p_duckdb_free(pointer);
+}
+static duckdb_state cth_duckdb_connect(duckdb_database database, duckdb_connection *out) {
+	return cth_p_duckdb_connect(database, out);
+}
+static duckdb_state cth_duckdb_query(duckdb_connection connection, const char *query, duckdb_result *out) {
+	return cth_p_duckdb_query(connection, query, out);
+}
+static const char *cth_duckdb_result_error(duckdb_result *result) {
+	return cth_p_duckdb_result_error(result);
+}
+static void cth_duckdb_destroy_result(duckdb_result *result) {
+	cth_p_duckdb_destroy_result(result);
+}
+static idx_t cth_duckdb_row_count(duckdb_result *result) {
+	return cth_p_duckdb_row_count(result);
+}
+static char *cth_duckdb_value_varchar(duckdb_result *result, idx_t column, idx_t row) {
+	return cth_p_duckdb_value_varchar(result, column, row);
+}
+static duckdb_state cth_duckdb_appender_create(duckdb_connection connection, const char *schema, const char *table, duckdb_appender *out) {
+	return cth_p_duckdb_appender_create(connection, schema, table, out);
+}
+static const char *cth_duckdb_appender_error(duckdb_appender appender) {
+	return cth_p_duckdb_appender_error(appender);
+}
+static duckdb_state cth_duckdb_appender_begin_row(duckdb_appender appender) {
+	return cth_p_duckdb_appender_begin_row(appender);
+}
+static duckdb_state cth_duckdb_append_int64(duckdb_appender appender, int64_t value) {
+	return cth_p_duckdb_append_int64(appender, value);
+}
+static duckdb_state cth_duckdb_append_varchar_length(duckdb_appender appender, const char *text, idx_t length) {
+	return cth_p_duckdb_append_varchar_length(appender, text, length);
+}
+static duckdb_value cth_duckdb_create_timestamp_tz(duckdb_timestamp input) {
+	return cth_p_duckdb_create_timestamp_tz(input);
+}
+static duckdb_state cth_duckdb_append_value(duckdb_appender appender, duckdb_value value) {
+	return cth_p_duckdb_append_value(appender, value);
+}
+static void cth_duckdb_destroy_value(duckdb_value *value) {
+	cth_p_duckdb_destroy_value(value);
+}
+static duckdb_state cth_duckdb_append_null(duckdb_appender appender) {
+	return cth_p_duckdb_append_null(appender);
+}
+static duckdb_state cth_duckdb_append_bool(duckdb_appender appender, bool value) {
+	return cth_p_duckdb_append_bool(appender, value);
+}
+static duckdb_state cth_duckdb_appender_end_row(duckdb_appender appender) {
+	return cth_p_duckdb_appender_end_row(appender);
+}
+static duckdb_state cth_duckdb_appender_flush(duckdb_appender appender) {
+	return cth_p_duckdb_appender_flush(appender);
+}
+static duckdb_state cth_duckdb_appender_close(duckdb_appender appender) {
+	return cth_p_duckdb_appender_close(appender);
+}
+static void cth_duckdb_appender_destroy(duckdb_appender *appender) {
+	cth_p_duckdb_appender_destroy(appender);
+}
+static void cth_duckdb_disconnect(duckdb_connection *connection) {
+	cth_p_duckdb_disconnect(connection);
+}
+static void cth_duckdb_close(duckdb_database *database) {
+	cth_p_duckdb_close(database);
+}
+*/
+import "C"
+
+import (
+	"crypto/rand"
+	"errors"
+	"fmt"
+	"io/fs"
+	"os"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+	"unsafe"
+)
+
+// This sink uses the system libduckdb C API directly instead of the duckdb-go
+// driver: duckdb-go bundles its own engine build, which can write a storage
+// format the system duckdb CLI (the tool agents query logs with) cannot
+// read, and it inflates the binary by tens of megabytes plus dozens of
+// module dependencies for a debug-only sink. The system library always
+// matches the same system's CLI, and loading it with dlopen keeps it an
+// optional runtime dependency: builds need the duckdb headers (hence the
+// duckdb build tag, mirroring brlapi), but the resulting reader starts and
+// runs normally on systems without libduckdb, and only --debug-db fails.
+
+// Rows are buffered in the appender only briefly: a flush happens at
+// duckDBFlushRows, on the first record at least duckDBFlushInterval after
+// the previous flush, and immediately for WARN and ERROR records so crash
+// diagnostics reach the database. A hard crash can still lose the rows
+// buffered since the last flush, unlike the file sink, which writes every
+// record straight away; the database otherwise trails live activity by
+// about a second, so it can be inspected (from a file copy, while the
+// reader holds the write lock) during a session.
+const (
+	duckDBFlushRows     = 512
+	duckDBFlushInterval = time.Second
+)
+
+var emptyDuckText = C.CString("")
+
+// Runtime packages install a versioned soname; the unversioned name is the
+// development symlink (present in Arch's duckdb package, absent from some
+// other distributions' runtime packages).
+var duckLibraryCandidates = []string{"libduckdb.so.1", "libduckdb.so"}
+
+var (
+	duckLoadOnce sync.Once
+	duckLoadErr  error
+)
+
+func loadDuckDBLibrary() error {
+	duckLoadOnce.Do(func() {
+		var failures []string
+		for _, candidate := range duckLibraryCandidates {
+			cname := C.CString(candidate)
+			message := C.cth_duckdb_load(cname)
+			C.free(unsafe.Pointer(cname))
+			if message == nil {
+				return
+			}
+			failures = append(failures, candidate+": "+C.GoString(message))
+		}
+		duckLoadErr = fmt.Errorf(
+			"%w: cannot load libduckdb (%s); install the duckdb package to use --debug-db",
+			ErrDuckDBUnavailable, strings.Join(failures, "; "))
+	})
+	return duckLoadErr
+}
+
+type duckDBWriter struct {
+	path      string
+	database  C.duckdb_database
+	conn      C.duckdb_connection
+	appender  C.duckdb_appender
+	sessionID string
+	seq       int64
+	pending   int
+	lastFlush time.Time
+
+	mutex     sync.Mutex
+	closeOnce sync.Once
+	closed    bool
+	writeErr  error
+	closeErr  error
+	cancelled atomic.Bool
+}
+
+// OpenDuckDB opens a DuckDB database at path as a diagnostics sink. Without
+// appendExisting, any existing database at path is replaced, mirroring
+// OpenFile. With it, existing sessions are kept and this run is recorded as
+// a new session row. The sink receives the same redacted NDJSON records as
+// the file sink, so redaction in Handler.format protects both.
+func OpenDuckDB(path string, appendExisting bool, pid int) (DiagnosticsWriter, error) {
+	if path == "" || path == "." || path == ".." {
+		return nil, errors.New("invalid diagnostics database path")
+	}
+	// Load before touching the filesystem: without the library we must not
+	// delete an existing database we then cannot replace.
+	if err := loadDuckDBLibrary(); err != nil {
+		return nil, err
+	}
+	if info, err := os.Stat(path); err == nil && info.IsDir() {
+		return nil, fmt.Errorf("diagnostics database path is a directory: %s", path)
+	}
+	if !appendExisting {
+		for _, target := range []string{path, path + ".wal"} {
+			if err := os.Remove(target); err != nil && !errors.Is(err, fs.ErrNotExist) {
+				return nil, fmt.Errorf("replace diagnostics database: %w", err)
+			}
+		}
+	}
+
+	sessionID, err := newSessionID()
+	if err != nil {
+		return nil, err
+	}
+	writer := &duckDBWriter{path: path, sessionID: sessionID}
+	if err := writer.open(pid); err != nil {
+		return nil, err
+	}
+	return writer, nil
+}
+
+func (writer *duckDBWriter) open(pid int) (err error) {
+	// Every failure past a successful open must release the handles already
+	// acquired and leave the writer in the closed state.
+	defer func() {
+		if err != nil {
+			writer.closed = true
+			writer.teardown()
+		}
+	}()
+
+	var config C.duckdb_config
+	if C.cth_duckdb_create_config(&config) == C.DuckDBError {
+		return errors.New("create duckdb config")
+	}
+	threads := C.CString("threads")
+	one := C.CString("1")
+	setState := C.cth_duckdb_set_config(config, threads, one)
+	C.free(unsafe.Pointer(threads))
+	C.free(unsafe.Pointer(one))
+	if setState == C.DuckDBError {
+		C.cth_duckdb_destroy_config(&config)
+		return errors.New("configure duckdb threads")
+	}
+
+	cpath := C.CString(writer.path)
+	var openErr *C.char
+	state := C.cth_duckdb_open_ext(cpath, &writer.database, config, &openErr)
+	C.free(unsafe.Pointer(cpath))
+	C.cth_duckdb_destroy_config(&config)
+	if state == C.DuckDBError {
+		message := C.GoString(openErr)
+		C.cth_duckdb_free(unsafe.Pointer(openErr))
+		return fmt.Errorf("open diagnostics database: %s", message)
+	}
+
+	// DuckDB creates the file with process umask permissions and rejects a
+	// pre-created empty file, so restrict it immediately: debug records can
+	// contain spoken text and the file sink creates its log at 0600.
+	if chmodErr := os.Chmod(writer.path, 0o600); chmodErr != nil {
+		return fmt.Errorf("restrict diagnostics database permissions: %w", chmodErr)
+	}
+
+	if C.cth_duckdb_connect(writer.database, &writer.conn) == C.DuckDBError {
+		return errors.New("connect diagnostics database")
+	}
+	for _, statement := range []string{duckDBCreateMeta, duckDBCreateSessions, duckDBCreateLogs} {
+		if execErr := duckExec(writer.conn, statement); execErr != nil {
+			return execErr
+		}
+	}
+	version, found, queryErr := duckQueryVarchar(writer.conn,
+		"SELECT value FROM meta WHERE key = 'schema_version'")
+	if queryErr != nil {
+		return queryErr
+	}
+	if found && version == duckDBSchemaVersionV1 {
+		// v1 stored every attribute in attrs only. Append the promoted
+		// columns (existing rows read NULL there; nothing is lost because
+		// attrs is complete) and restamp the version.
+		for _, attr := range promotedAttrs {
+			statement := fmt.Sprintf("ALTER TABLE logs ADD COLUMN IF NOT EXISTS \"%s\" %s", attr.key, attr.sqlType())
+			if execErr := duckExec(writer.conn, statement); execErr != nil {
+				return execErr
+			}
+		}
+		if execErr := duckExec(writer.conn,
+			"UPDATE meta SET value = '"+duckDBSchemaVersion+"' WHERE key = 'schema_version'"); execErr != nil {
+			return execErr
+		}
+	} else if found && version != duckDBSchemaVersion {
+		return fmt.Errorf("unsupported diagnostics database schema version %q (want %q); use a new file or omit --append",
+			version, duckDBSchemaVersion)
+	}
+	if !found {
+		for _, statement := range []string{
+			"INSERT INTO meta VALUES ('format', 'cthulhu-debug-log') ON CONFLICT DO NOTHING",
+			"INSERT INTO meta VALUES ('schema_version', '" + duckDBSchemaVersion + "') ON CONFLICT DO NOTHING",
+		} {
+			if execErr := duckExec(writer.conn, statement); execErr != nil {
+				return execErr
+			}
+		}
+	}
+	// sessionID is generated by newSessionID from random bytes (hex and
+	// dashes only) and the timestamp is RFC3339, so both are safe SQL
+	// literal content.
+	insertSession := fmt.Sprintf(
+		"INSERT INTO sessions VALUES ('%s', TIMESTAMPTZ '%s', %d)",
+		writer.sessionID, time.Now().UTC().Format(time.RFC3339Nano), pid)
+	if execErr := duckExec(writer.conn, insertSession); execErr != nil {
+		return execErr
+	}
+
+	table := C.CString("logs")
+	state = C.cth_duckdb_appender_create(writer.conn, nil, table, &writer.appender)
+	C.free(unsafe.Pointer(table))
+	if state == C.DuckDBError {
+		return fmt.Errorf("create diagnostics appender: %s", duckAppenderError(writer.appender))
+	}
+	writer.lastFlush = time.Now()
+	// Schema and session setup already wrote through the WAL; restrict it
+	// now rather than waiting for the first flush.
+	_ = os.Chmod(writer.path+".wal", 0o600)
+	return nil
+}
+
+func (writer *duckDBWriter) Write(data []byte) (int, error) {
+	writer.mutex.Lock()
+	defer writer.mutex.Unlock()
+	if writer.closed || writer.cancelled.Load() {
+		return 0, fmt.Errorf("write diagnostics database: %w", fs.ErrClosed)
+	}
+	if writer.writeErr != nil {
+		return 0, writer.writeErr
+	}
+	record, ok := parseLogRecord(data)
+	if !ok {
+		// The formatter is the only producer, so this should not happen;
+		// keep the raw record rather than silently dropping it.
+		record = parsedRecord{
+			timestamp: time.Now(),
+			level:     "UNKNOWN",
+			message:   strings.TrimRight(string(data), "\n"),
+			attrs:     "{}",
+		}
+	}
+	if err := writer.appendRecord(record); err != nil {
+		writer.writeErr = err
+		return 0, err
+	}
+	writer.pending++
+	if writer.pending >= duckDBFlushRows || record.level == "WARN" || record.level == "ERROR" ||
+		time.Since(writer.lastFlush) >= duckDBFlushInterval {
+		if err := writer.flushLocked(); err != nil {
+			writer.writeErr = err
+			return 0, err
+		}
+	}
+	return len(data), nil
+}
+
+// appendRecord appends one row in logs table column order.
+func (writer *duckDBWriter) appendRecord(record parsedRecord) error {
+	writer.seq++
+	app := writer.appender
+	if C.cth_duckdb_appender_begin_row(app) == C.DuckDBError {
+		return writer.appendError("begin row")
+	}
+	if C.cth_duckdb_append_int64(app, C.int64_t(writer.seq)) == C.DuckDBError {
+		return writer.appendError("append seq")
+	}
+	if duckAppendText(app, writer.sessionID) == C.DuckDBError {
+		return writer.appendError("append session_id")
+	}
+	timestamp := C.duckdb_timestamp{micros: C.int64_t(record.timestamp.UnixMicro())}
+	value := C.cth_duckdb_create_timestamp_tz(timestamp)
+	state := C.cth_duckdb_append_value(app, value)
+	C.cth_duckdb_destroy_value(&value)
+	if state == C.DuckDBError {
+		return writer.appendError("append ts")
+	}
+	for _, field := range []struct {
+		name string
+		text string
+	}{
+		{"level", record.level},
+		{"message", record.message},
+	} {
+		if duckAppendText(app, field.text) == C.DuckDBError {
+			return writer.appendError("append " + field.name)
+		}
+	}
+	for _, field := range []struct {
+		name string
+		text string
+	}{
+		{"source", record.source},
+		{"function", record.function},
+	} {
+		if field.text == "" {
+			if C.cth_duckdb_append_null(app) == C.DuckDBError {
+				return writer.appendError("append " + field.name)
+			}
+			continue
+		}
+		if duckAppendText(app, field.text) == C.DuckDBError {
+			return writer.appendError("append " + field.name)
+		}
+	}
+	if duckAppendText(app, record.attrs) == C.DuckDBError {
+		return writer.appendError("append attrs")
+	}
+	for i, attr := range promotedAttrs {
+		var value promotedValue
+		if i < len(record.promoted) {
+			value = record.promoted[i]
+		}
+		if !value.valid {
+			if C.cth_duckdb_append_null(app) == C.DuckDBError {
+				return writer.appendError("append " + attr.key)
+			}
+			continue
+		}
+		var state C.duckdb_state
+		switch attr.kind {
+		case promotedText:
+			state = duckAppendText(app, value.text)
+		case promotedInt:
+			state = C.cth_duckdb_append_int64(app, C.int64_t(value.integer))
+		case promotedBool:
+			state = C.cth_duckdb_append_bool(app, C.bool(value.boolean))
+		}
+		if state == C.DuckDBError {
+			return writer.appendError("append " + attr.key)
+		}
+	}
+	if C.cth_duckdb_appender_end_row(app) == C.DuckDBError {
+		return writer.appendError("end row")
+	}
+	return nil
+}
+
+func (writer *duckDBWriter) appendError(step string) error {
+	return fmt.Errorf("write diagnostics database (%s): %s", step, duckAppenderError(writer.appender))
+}
+
+func (writer *duckDBWriter) flushLocked() error {
+	if C.cth_duckdb_appender_flush(writer.appender) == C.DuckDBError {
+		return fmt.Errorf("flush diagnostics database: %s", duckAppenderError(writer.appender))
+	}
+	writer.pending = 0
+	writer.lastFlush = time.Now()
+	// The write-ahead log holds recent records until checkpoint; restrict it
+	// too whenever it exists. Best effort: DuckDB may not have created it.
+	_ = os.Chmod(writer.path+".wal", 0o600)
+	return nil
+}
+
+// Cancel flags the sink cancelled and returns immediately. Diagnostics.Stop
+// calls Cancel synchronously while enforcing its deadline, so it must never
+// wait for an in-flight appender call or run the closing flush/checkpoint;
+// that teardown belongs to Close, which the diagnostics writer goroutine
+// calls after its queue drains. This mirrors fileWriter, whose Cancel is a
+// bare file close that never waits for an in-flight Write.
+func (writer *duckDBWriter) Cancel() error {
+	writer.cancelled.Store(true)
+	return nil
+}
+
+func (writer *duckDBWriter) Close() error {
+	writer.closeOnce.Do(func() {
+		writer.mutex.Lock()
+		defer writer.mutex.Unlock()
+		writer.closed = true
+		if writer.appender != nil && C.cth_duckdb_appender_close(writer.appender) == C.DuckDBError {
+			writer.closeErr = fmt.Errorf("close diagnostics appender: %s",
+				duckAppenderError(writer.appender))
+		}
+		writer.teardown()
+	})
+	return writer.closeErr
+}
+
+// teardown releases any handles still held, in reverse acquisition order.
+// Callers outside open must hold the mutex.
+func (writer *duckDBWriter) teardown() {
+	if writer.appender != nil {
+		C.cth_duckdb_appender_destroy(&writer.appender)
+	}
+	if writer.conn != nil {
+		C.cth_duckdb_disconnect(&writer.conn)
+	}
+	if writer.database != nil {
+		C.cth_duckdb_close(&writer.database)
+	}
+}
+
+func duckExec(conn C.duckdb_connection, sql string) error {
+	csql := C.CString(sql)
+	defer C.free(unsafe.Pointer(csql))
+	var result C.duckdb_result
+	if C.cth_duckdb_query(conn, csql, &result) == C.DuckDBError {
+		message := C.GoString(C.cth_duckdb_result_error(&result))
+		C.cth_duckdb_destroy_result(&result)
+		return fmt.Errorf("diagnostics database statement: %s", message)
+	}
+	C.cth_duckdb_destroy_result(&result)
+	return nil
+}
+
+func duckQueryVarchar(conn C.duckdb_connection, sql string) (string, bool, error) {
+	csql := C.CString(sql)
+	defer C.free(unsafe.Pointer(csql))
+	var result C.duckdb_result
+	if C.cth_duckdb_query(conn, csql, &result) == C.DuckDBError {
+		message := C.GoString(C.cth_duckdb_result_error(&result))
+		C.cth_duckdb_destroy_result(&result)
+		return "", false, fmt.Errorf("diagnostics database query: %s", message)
+	}
+	defer C.cth_duckdb_destroy_result(&result)
+	if C.cth_duckdb_row_count(&result) == 0 {
+		return "", false, nil
+	}
+	text := C.cth_duckdb_value_varchar(&result, 0, 0)
+	value := C.GoString(text)
+	C.cth_duckdb_free(unsafe.Pointer(text))
+	return value, true, nil
+}
+
+func duckAppendText(app C.duckdb_appender, text string) C.duckdb_state {
+	if len(text) == 0 {
+		return C.cth_duckdb_append_varchar_length(app, emptyDuckText, 0)
+	}
+	// The length variant preserves embedded NUL bytes that JSON decoding can
+	// legitimately produce ("\u0000"); the NUL-terminated variant truncates.
+	state := C.cth_duckdb_append_varchar_length(
+		app, (*C.char)(unsafe.Pointer(unsafe.StringData(text))), C.idx_t(len(text)))
+	// StringData must stay live across the call above.
+	_ = text
+	return state
+}
+
+func duckAppenderError(app C.duckdb_appender) string {
+	if app == nil {
+		return "unknown error"
+	}
+	return C.GoString(C.cth_duckdb_appender_error(app))
+}
+
+func newSessionID() (string, error) {
+	var raw [16]byte
+	if _, err := rand.Read(raw[:]); err != nil {
+		return "", fmt.Errorf("generate diagnostics session id: %w", err)
+	}
+	raw[6] = raw[6]&0x0f | 0x40
+	raw[8] = raw[8]&0x3f | 0x80
+	return fmt.Sprintf("%x-%x-%x-%x-%x", raw[0:4], raw[4:6], raw[6:8], raw[8:10], raw[10:16]), nil
+}

internal/logging/duckdb_native_test.go

@@ -0,0 +1,318 @@
+//go:build duckdb && cgo
+
+// Cthulhu screen reader.
+// Copyright (c) 2026 Storm Dragon
+// Authors: Storm Dragon, Angus Kola
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free
+// Software Foundation, either version 3 of the License, or (at your option)
+// any later version.
+//
+// This program is distributed in the hope that it will be useful, but WITHOUT
+// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+// FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+// more details.
+//
+// You should have received a copy of the GNU General Public License along with
+// this program. If not, see <https://www.gnu.org/licenses/gpl-3.0.html>.
+
+package logging
+
+import (
+	"errors"
+	"io/fs"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+)
+
+const (
+	duckTestLineInfo = `{"message":"starting Cthulhu","level":"INFO","source":"/srv/cthulhu/cmd/cthulhu/main.go:201","function":"main.run","pid":42,"time":"2026-11-04T03:58:47.316096257-05:00"}` + "\n"
+	duckTestLineDup  = `{"message":"document navigation completed","level":"DEBUG","source":"/srv/cthulhu/internal/app/document_review.go:319","function":"example.present","level":3,"role":"heading","time":"2026-11-04T04:01:02.5-05:00"}` + "\n"
+	duckTestLineErr  = `{"message":"speech delivery failed","level":"ERROR","source":"/srv/cthulhu/internal/app/speech.go:9","function":"example.deliver","error":"socket closed","time":"2026-11-04T04:02:03-05:00"}` + "\n"
+)
+
+func duckQuery(t *testing.T, path, query string) string {
+	t.Helper()
+	cli, err := exec.LookPath("duckdb")
+	if err != nil {
+		t.Skip("duckdb CLI not installed; cannot verify queryability")
+	}
+	output, err := exec.Command(cli, "-readonly", "-noheader", "-list", path, "-c", query).CombinedOutput()
+	if err != nil {
+		t.Fatalf("duckdb query %q: %v\n%s", query, err, output)
+	}
+	return strings.TrimSpace(string(output))
+}
+
+func TestDuckDBSinkStoresQueryableRecords(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "debug.duckdb")
+	writer, err := OpenDuckDB(path, false, 4242)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, line := range []string{duckTestLineInfo, duckTestLineDup, duckTestLineErr, "not a formatted record\n"} {
+		if _, err := writer.Write([]byte(line)); err != nil {
+			t.Fatalf("Write %q: %v", line, err)
+		}
+	}
+	if err := writer.Close(); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := writer.Write([]byte(duckTestLineInfo)); !errors.Is(err, fs.ErrClosed) {
+		t.Fatalf("write after close error = %v, want fs.ErrClosed", err)
+	}
+	info, err := os.Stat(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if info.Mode().Perm() != 0o600 {
+		t.Fatalf("database permissions = %o, want 600", info.Mode().Perm())
+	}
+
+	if got := duckQuery(t, path, "SELECT count(*) FROM logs"); got != "4" {
+		t.Fatalf("log rows = %s, want 4", got)
+	}
+	if got := duckQuery(t, path,
+		"SELECT level FROM logs WHERE message = 'document navigation completed'"); got != "DEBUG" {
+		t.Fatalf("typed level = %s, want DEBUG", got)
+	}
+	if got := duckQuery(t, path,
+		"SELECT attrs->>'$.level' FROM logs WHERE message = 'document navigation completed'"); got != "3" {
+		t.Fatalf("attrs level = %s, want 3", got)
+	}
+	if got := duckQuery(t, path,
+		"SELECT attrs->>'$.error' FROM logs WHERE level = 'ERROR'"); got != "socket closed" {
+		t.Fatalf("attrs error = %s", got)
+	}
+	// The unformatted line must be preserved, not dropped.
+	if got := duckQuery(t, path, "SELECT message FROM logs WHERE level = 'UNKNOWN'"); got != "not a formatted record" {
+		t.Fatalf("fallback message = %q", got)
+	}
+	if got := duckQuery(t, path,
+		"SELECT count(DISTINCT session_id) || '|' || min(seq) || '|' || max(seq) FROM logs"); got != "1|1|4" {
+		t.Fatalf("session/seq = %s, want 1|1|4", got)
+	}
+	if got := duckQuery(t, path, "SELECT pid FROM sessions"); got != "4242" {
+		t.Fatalf("session pid = %s", got)
+	}
+	if got := duckQuery(t, path, "SELECT value FROM meta WHERE key = 'schema_version'"); got != duckDBSchemaVersion {
+		t.Fatalf("schema_version = %s", got)
+	}
+	// Timestamps are real TIMESTAMPTZ values that support range filters.
+	if got := duckQuery(t, path,
+		"SELECT count(*) FROM logs WHERE level != 'UNKNOWN' AND ts >= TIMESTAMPTZ '2026-11-04 04:00:00-05:00'"); got != "2" {
+		t.Fatalf("range-filtered rows = %s, want 2", got)
+	}
+}
+
+func TestDuckDBSinkAppendsSessionsAndReplaces(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "debug.duckdb")
+	write := func(appendExisting bool, line string) {
+		t.Helper()
+		writer, err := OpenDuckDB(path, appendExisting, 7)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, err := writer.Write([]byte(line)); err != nil {
+			t.Fatal(err)
+		}
+		if err := writer.Close(); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	write(false, duckTestLineInfo)
+	write(true, duckTestLineErr)
+	if got := duckQuery(t, path, "SELECT count(*) || '|' || count(DISTINCT session_id) FROM logs"); got != "2|2" {
+		t.Fatalf("after append rows|sessions = %s, want 2|2", got)
+	}
+	// Each session restarts its own sequence.
+	if got := duckQuery(t, path, "SELECT count(*) FROM logs WHERE seq = 1"); got != "2" {
+		t.Fatalf("seq=1 rows = %s, want 2", got)
+	}
+
+	write(false, duckTestLineDup)
+	if got := duckQuery(t, path, "SELECT count(*) || '|' || count(DISTINCT session_id) FROM logs"); got != "1|1" {
+		t.Fatalf("after replace rows|sessions = %s, want 1|1", got)
+	}
+}
+
+func TestDuckDBSinkRejectsDirectoryAndForeignFile(t *testing.T) {
+	if _, err := OpenDuckDB(t.TempDir(), false, 1); err == nil {
+		t.Fatal("directory path accepted")
+	}
+	foreign := filepath.Join(t.TempDir(), "foreign.duckdb")
+	if err := os.WriteFile(foreign, []byte("not a duckdb database"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := OpenDuckDB(foreign, true, 1); err == nil {
+		t.Fatal("foreign file accepted in append mode")
+	}
+	// Replace mode discards the foreign file, mirroring OpenFile truncation.
+	writer, err := OpenDuckDB(foreign, false, 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := writer.Close(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// At idle record rates the row-count threshold alone would leave the
+// database empty for minutes; a record arriving after the flush interval
+// must flush everything buffered so far.
+func TestDuckDBSinkFlushesAfterInterval(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "debug.duckdb")
+	sink, err := OpenDuckDB(path, false, 7)
+	if err != nil {
+		t.Fatal(err)
+	}
+	writer := sink.(*duckDBWriter)
+	if wal, err := os.Stat(path + ".wal"); err == nil && wal.Mode().Perm() != 0o600 {
+		t.Fatalf("wal permissions at open = %o, want 600", wal.Mode().Perm())
+	}
+	if _, err := writer.Write([]byte(duckTestLineInfo)); err != nil {
+		t.Fatal(err)
+	}
+	if writer.pending != 1 {
+		t.Fatalf("pending after one INFO record = %d, want 1 (buffered)", writer.pending)
+	}
+	writer.lastFlush = time.Now().Add(-2 * duckDBFlushInterval)
+	if _, err := writer.Write([]byte(duckTestLineDup)); err != nil {
+		t.Fatal(err)
+	}
+	if writer.pending != 0 {
+		t.Fatalf("pending after interval record = %d, want 0 (flushed)", writer.pending)
+	}
+	if err := writer.Close(); err != nil {
+		t.Fatal(err)
+	}
+	if got := duckQuery(t, path, "SELECT count(*) FROM logs"); got != "2" {
+		t.Fatalf("log rows = %s, want 2", got)
+	}
+}
+
+func duckExecCLI(t *testing.T, path, statement string) {
+	t.Helper()
+	cli, err := exec.LookPath("duckdb")
+	if err != nil {
+		t.Skip("duckdb CLI not installed; cannot prepare fixture database")
+	}
+	if output, err := exec.Command(cli, path, "-c", statement).CombinedOutput(); err != nil {
+		t.Fatalf("duckdb exec %q: %v\n%s", statement, err, output)
+	}
+}
+
+// Promoted attributes must land in typed columns (queryable without JSON
+// extraction or casts) while remaining in attrs.
+func TestDuckDBSinkPromotedColumns(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "debug.duckdb")
+	writer, err := OpenDuckDB(path, false, 7)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, line := range []string{duckTestLineInfo, duckTestLineDup, duckTestLineErr} {
+		if _, err := writer.Write([]byte(line)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := writer.Close(); err != nil {
+		t.Fatal(err)
+	}
+	if got := duckQuery(t, path,
+		"SELECT role FROM logs WHERE message = 'document navigation completed'"); got != "heading" {
+		t.Fatalf("promoted role = %s, want heading", got)
+	}
+	if got := duckQuery(t, path,
+		"SELECT error FROM logs WHERE level = 'ERROR'"); got != "socket closed" {
+		t.Fatalf("promoted error = %s, want socket closed", got)
+	}
+	if got := duckQuery(t, path,
+		"SELECT count(*) FROM logs WHERE role IS NULL"); got != "2" {
+		t.Fatalf("null-role rows = %s, want 2", got)
+	}
+	// Column types are honest SQL types, not text.
+	if got := duckQuery(t, path,
+		"SELECT data_type FROM information_schema.columns WHERE table_name = 'logs' AND column_name = 'snapshot_duration'"); got != "BIGINT" {
+		t.Fatalf("snapshot_duration type = %s, want BIGINT", got)
+	}
+	if got := duckQuery(t, path,
+		"SELECT data_type FROM information_schema.columns WHERE table_name = 'logs' AND column_name = 'focused'"); got != "BOOLEAN" {
+		t.Fatalf("focused type = %s, want BOOLEAN", got)
+	}
+}
+
+// A v1 database (envelope columns and attrs only) must upgrade in place on
+// append: existing rows keep their attrs and read NULL in the new columns.
+func TestDuckDBSinkUpgradesV1Database(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "debug.duckdb")
+	duckExecCLI(t, path, `
+		CREATE TABLE meta (key VARCHAR PRIMARY KEY, value VARCHAR NOT NULL);
+		INSERT INTO meta VALUES ('format', 'cthulhu-debug-log'), ('schema_version', '1');
+		CREATE TABLE sessions (session_id VARCHAR PRIMARY KEY, started_at TIMESTAMPTZ NOT NULL, pid BIGINT NOT NULL);
+		INSERT INTO sessions VALUES ('old-session', TIMESTAMPTZ '2026-11-04 03:58:47-05:00', 99);
+		CREATE TABLE logs (seq BIGINT NOT NULL, session_id VARCHAR NOT NULL, ts TIMESTAMPTZ NOT NULL,
+			level VARCHAR NOT NULL, message VARCHAR NOT NULL, source VARCHAR, function VARCHAR, attrs JSON NOT NULL);
+		INSERT INTO logs VALUES (1, 'old-session', TIMESTAMPTZ '2026-11-04 03:58:47-05:00',
+			'DEBUG', 'document navigation completed', NULL, NULL, '{"role":"heading"}');
+	`)
+	writer, err := OpenDuckDB(path, true, 7)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := writer.Write([]byte(duckTestLineDup)); err != nil {
+		t.Fatal(err)
+	}
+	if err := writer.Close(); err != nil {
+		t.Fatal(err)
+	}
+	if got := duckQuery(t, path, "SELECT value FROM meta WHERE key = 'schema_version'"); got != duckDBSchemaVersion {
+		t.Fatalf("schema_version after upgrade = %s, want %s", got, duckDBSchemaVersion)
+	}
+	if got := duckQuery(t, path,
+		"SELECT role || '|' || (attrs->>'$.role') FROM logs WHERE session_id != 'old-session'"); got != "heading|heading" {
+		t.Fatalf("new row role column|attrs = %s", got)
+	}
+	if got := duckQuery(t, path,
+		"SELECT (role IS NULL) || '|' || (attrs->>'$.role') FROM logs WHERE session_id = 'old-session'"); got != "true|heading" {
+		t.Fatalf("old row role-null|attrs = %s, want true|heading", got)
+	}
+}
+
+// Cancel is called synchronously by Diagnostics.Stop while it enforces its
+// deadline, so it must only flag cancellation and return at once; records
+// already appended are still flushed by the Close that follows.
+func TestDuckDBSinkCancelIsNonBlocking(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "debug.duckdb")
+	writer, err := OpenDuckDB(path, false, 7)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := writer.Write([]byte(duckTestLineInfo)); err != nil {
+		t.Fatal(err)
+	}
+	done := make(chan error, 1)
+	go func() { done <- writer.Cancel() }()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatal(err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("Cancel blocked")
+	}
+	if _, err := writer.Write([]byte(duckTestLineDup)); !errors.Is(err, fs.ErrClosed) {
+		t.Fatalf("write after cancel error = %v, want fs.ErrClosed", err)
+	}
+	if err := writer.Close(); err != nil {
+		t.Fatal(err)
+	}
+	if got := duckQuery(t, path, "SELECT count(*) FROM logs"); got != "1" {
+		t.Fatalf("log rows = %s, want 1 (pre-cancel record flushed by Close)", got)
+	}
+}

internal/logging/duckdb_test.go

@@ -0,0 +1,228 @@
+// Cthulhu screen reader.
+// Copyright (c) 2026 Storm Dragon
+// Authors: Storm Dragon, Angus Kola
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free
+// Software Foundation, either version 3 of the License, or (at your option)
+// any later version.
+//
+// This program is distributed in the hope that it will be useful, but WITHOUT
+// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+// FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+// more details.
+//
+// You should have received a copy of the GNU General Public License along with
+// this program. If not, see <https://www.gnu.org/licenses/gpl-3.0.html>.
+
+package logging
+
+import (
+	"encoding/json"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestParseLogRecordSplitsTypedColumnsAndAttrs(t *testing.T) {
+	line := []byte(`{"message":"starting Cthulhu","level":"INFO","source":"/srv/cthulhu/cmd/cthulhu/main.go:201","function":"main.run","pid":259626,"build":{"go_version":"go1.27"},"x11_display_available":true,"time":"2026-11-04T03:58:47.316096257-05:00"}` + "\n")
+
+	record, ok := parseLogRecord(line)
+	if !ok {
+		t.Fatal("record did not parse")
+	}
+	if record.message != "starting Cthulhu" || record.level != "INFO" {
+		t.Fatalf("message/level = %q/%q", record.message, record.level)
+	}
+	if record.source != "/srv/cthulhu/cmd/cthulhu/main.go:201" || record.function != "main.run" {
+		t.Fatalf("source/function = %q/%q", record.source, record.function)
+	}
+	wantTime, err := time.Parse(time.RFC3339Nano, "2026-11-04T03:58:47.316096257-05:00")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !record.timestamp.Equal(wantTime) {
+		t.Fatalf("timestamp = %v, want %v", record.timestamp, wantTime)
+	}
+	var attrs map[string]json.RawMessage
+	if err := json.Unmarshal([]byte(record.attrs), &attrs); err != nil {
+		t.Fatalf("attrs is not a JSON object: %v (%s)", err, record.attrs)
+	}
+	if string(attrs["pid"]) != "259626" || string(attrs["build"]) != `{"go_version":"go1.27"}` ||
+		string(attrs["x11_display_available"]) != "true" {
+		t.Fatalf("attrs = %s", record.attrs)
+	}
+	for _, reserved := range []string{"message", "level", "source", "function", "time"} {
+		if _, present := attrs[reserved]; present {
+			t.Fatalf("reserved key %q leaked into attrs: %s", reserved, record.attrs)
+		}
+	}
+}
+
+// Some records carry an attribute that reuses a reserved name with a
+// different type (an integer heading "level", for example). The formatter
+// emits its own level first, so the typed column keeps the string level and
+// the attribute lands in attrs.
+func TestParseLogRecordDuplicateReservedAttrGoesToAttrs(t *testing.T) {
+	line := []byte(`{"message":"document navigation completed","level":"DEBUG","source":"/srv/cthulhu/internal/app/document_review.go:319","function":"example.present","level":3,"role":"heading","time":"2026-11-04T04:01:02.5-05:00"}` + "\n")
+
+	record, ok := parseLogRecord(line)
+	if !ok {
+		t.Fatal("record did not parse")
+	}
+	if record.level != "DEBUG" {
+		t.Fatalf("level = %q, want DEBUG", record.level)
+	}
+	var attrs map[string]json.RawMessage
+	if err := json.Unmarshal([]byte(record.attrs), &attrs); err != nil {
+		t.Fatal(err)
+	}
+	if string(attrs["level"]) != "3" || string(attrs["role"]) != `"heading"` {
+		t.Fatalf("attrs = %s", record.attrs)
+	}
+}
+
+func TestParseLogRecordRejectsMalformedLines(t *testing.T) {
+	for name, line := range map[string]string{
+		"empty":            "",
+		"not json":         "starting Cthulhu\n",
+		"array":            `[{"message":"m"}]` + "\n",
+		"missing time":     `{"message":"m","level":"INFO"}` + "\n",
+		"missing message":  `{"level":"INFO","time":"2026-11-04T03:58:47Z"}` + "\n",
+		"numeric level":    `{"message":"m","level":1,"time":"2026-11-04T03:58:47Z"}` + "\n",
+		"bad time":         `{"message":"m","level":"INFO","time":"not-a-time"}` + "\n",
+		"trailing garbage": `{"message":"m","level":"INFO","time":"2026-11-04T03:58:47Z"} extra` + "\n",
+		"unclosed":         `{"message":"m","level":"INFO"` + "\n",
+	} {
+		if _, ok := parseLogRecord([]byte(line)); ok {
+			t.Errorf("%s: parsed unexpectedly", name)
+		}
+	}
+}
+
+func promotedByKey(record parsedRecord, key string) promotedValue {
+	for i, attr := range promotedAttrs {
+		if attr.key == key && i < len(record.promoted) {
+			return record.promoted[i]
+		}
+	}
+	return promotedValue{}
+}
+
+func TestParseLogRecordPromotesTypedAttrs(t *testing.T) {
+	line := []byte(`{"message":"received accessibility event","level":"DEBUG","source":"s:1","function":"f","role":"heading","snapshot_duration":12345,"focused":true,"error":null,"selection_start":"3","custom_thing":{"nested":1},"time":"2026-11-04T03:58:47Z"}` + "\n")
+
+	record, ok := parseLogRecord(line)
+	if !ok {
+		t.Fatal("record did not parse")
+	}
+	if got := promotedByKey(record, "role"); !got.valid || got.text != "heading" {
+		t.Fatalf("role = %+v", got)
+	}
+	if got := promotedByKey(record, "snapshot_duration"); !got.valid || got.integer != 12345 {
+		t.Fatalf("snapshot_duration = %+v", got)
+	}
+	if got := promotedByKey(record, "focused"); !got.valid || !got.boolean {
+		t.Fatalf("focused = %+v", got)
+	}
+	// JSON null and wrong-typed values stay out of the typed columns but
+	// remain in attrs.
+	if got := promotedByKey(record, "error"); got.valid {
+		t.Fatalf("error = %+v, want invalid for null", got)
+	}
+	if got := promotedByKey(record, "selection_start"); got.valid {
+		t.Fatalf("selection_start = %+v, want invalid for string in int column", got)
+	}
+	if got := promotedByKey(record, "application"); got.valid {
+		t.Fatalf("application = %+v, want invalid when absent", got)
+	}
+	var attrs map[string]json.RawMessage
+	if err := json.Unmarshal([]byte(record.attrs), &attrs); err != nil {
+		t.Fatal(err)
+	}
+	if string(attrs["selection_start"]) != `"3"` || string(attrs["custom_thing"]) != `{"nested":1}` ||
+		string(attrs["role"]) != `"heading"` {
+		t.Fatalf("attrs lost data: %s", record.attrs)
+	}
+}
+
+// The formatter emits its timestamp last, after record attributes. An
+// attribute also named "time" must land in attrs and never steal the ts
+// column or sink the whole record to the fallback row.
+func TestParseLogRecordTimeAttrDoesNotStealTimestamp(t *testing.T) {
+	for name, attr := range map[string]string{
+		"numeric attr": `123`,
+		"rfc3339 attr": `"1999-01-01T00:00:00Z"`,
+		"garbage attr": `"not-a-time"`,
+	} {
+		line := []byte(`{"message":"m","level":"INFO","source":"s:1","function":"f","time":` + attr +
+			`,"role":"heading","time":"2026-11-04T03:58:47Z"}` + "\n")
+		record, ok := parseLogRecord(line)
+		if !ok {
+			t.Fatalf("%s: record did not parse", name)
+		}
+		want, err := time.Parse(time.RFC3339Nano, "2026-11-04T03:58:47Z")
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !record.timestamp.Equal(want) {
+			t.Fatalf("%s: timestamp = %v, want %v", name, record.timestamp, want)
+		}
+		var attrs map[string]json.RawMessage
+		if err := json.Unmarshal([]byte(record.attrs), &attrs); err != nil {
+			t.Fatal(err)
+		}
+		if string(attrs["time"]) != attr {
+			t.Fatalf("%s: attrs time = %s, want %s", name, attrs["time"], attr)
+		}
+		if got := promotedByKey(record, "role"); !got.valid || got.text != "heading" {
+			t.Fatalf("%s: promoted role = %+v", name, got)
+		}
+	}
+}
+
+type recordingSink struct {
+	writes   [][]byte
+	writeErr error
+	cancels  int
+	closes   int
+}
+
+func (sink *recordingSink) Write(data []byte) (int, error) {
+	sink.writes = append(sink.writes, append([]byte(nil), data...))
+	return len(data), sink.writeErr
+}
+
+func (sink *recordingSink) Cancel() error { sink.cancels++; return nil }
+func (sink *recordingSink) Close() error  { sink.closes++; return nil }
+
+func TestMultiWriterFansOutDespiteFailure(t *testing.T) {
+	failing := &recordingSink{writeErr: errors.New("sink failed")}
+	healthy := &recordingSink{}
+	writer := NewMultiWriter(failing, healthy)
+
+	line := []byte("{\"message\":\"m\"}\n")
+	if _, err := writer.Write(line); err == nil || !strings.Contains(err.Error(), "sink failed") {
+		t.Fatalf("Write error = %v", err)
+	}
+	if len(failing.writes) != 1 || len(healthy.writes) != 1 {
+		t.Fatalf("writes = %d/%d, want both sinks served", len(failing.writes), len(healthy.writes))
+	}
+	if err := writer.Close(); err != nil {
+		t.Fatal(err)
+	}
+	if failing.closes != 1 || healthy.closes != 1 {
+		t.Fatalf("closes = %d/%d", failing.closes, healthy.closes)
+	}
+}
+
+func TestMultiWriterPassesThroughSingleSink(t *testing.T) {
+	sink := &recordingSink{}
+	if got := NewMultiWriter(sink); got != DiagnosticsWriter(sink) {
+		t.Fatalf("single sink wrapped: %T", got)
+	}
+	if got := NewMultiWriter(); got == nil {
+		t.Fatal("empty multi writer is nil")
+	}
+}

internal/logging/duckdb_unavailable.go

@@ -0,0 +1,25 @@
+//go:build !duckdb || !cgo
+
+// Cthulhu screen reader.
+// Copyright (c) 2026 Storm Dragon
+// Authors: Storm Dragon, Angus Kola
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free
+// Software Foundation, either version 3 of the License, or (at your option)
+// any later version.
+//
+// This program is distributed in the hope that it will be useful, but WITHOUT
+// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+// FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+// more details.
+//
+// You should have received a copy of the GNU General Public License along with
+// this program. If not, see <https://www.gnu.org/licenses/gpl-3.0.html>.
+
+package logging
+
+// OpenDuckDB is provided by duckdb_native.go in builds with -tags duckdb.
+func OpenDuckDB(_ string, _ bool, _ int) (DiagnosticsWriter, error) {
+	return nil, ErrDuckDBUnavailable
+}

internal/logging/duckdb_unavailable_test.go

@@ -0,0 +1,36 @@
+//go:build !duckdb || !cgo
+
+// Cthulhu screen reader.
+// Copyright (c) 2026 Storm Dragon
+// Authors: Storm Dragon, Angus Kola
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free
+// Software Foundation, either version 3 of the License, or (at your option)
+// any later version.
+//
+// This program is distributed in the hope that it will be useful, but WITHOUT
+// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+// FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+// more details.
+//
+// You should have received a copy of the GNU General Public License along with
+// this program. If not, see <https://www.gnu.org/licenses/gpl-3.0.html>.
+
+package logging
+
+import (
+	"errors"
+	"path/filepath"
+	"testing"
+)
+
+func TestOpenDuckDBUnavailableWithoutBuildTag(t *testing.T) {
+	writer, err := OpenDuckDB(filepath.Join(t.TempDir(), "debug.duckdb"), false, 1)
+	if writer != nil {
+		t.Fatalf("writer = %v, want nil", writer)
+	}
+	if !errors.Is(err, ErrDuckDBUnavailable) {
+		t.Fatalf("error = %v, want ErrDuckDBUnavailable", err)
+	}
+}