diff --git a/docs/gossip-protocol.md b/docs/gossip-protocol.md index 83f6b88..0e9a6ee 100644 --- a/docs/gossip-protocol.md +++ b/docs/gossip-protocol.md @@ -213,14 +213,17 @@ truth. ## 8. Milestones -**M1 — Determinism fixes (prereq).** F1 (canonical fingerprints), F2 (HLC in all -"when" columns), F3 (TF-IDF tie-break), F4 (locator IDs + `ORDER BY hcl`). +**M1 — Determinism fixes (prereq).** DONE. F1 (canonical fingerprints), F2 (HLC +in all "when" columns), F3 (TF-IDF tie-break), F4 (locator IDs + `ORDER BY hcl`). Verify: two fresh DBs ingesting the same real content produce identical observation hashes and identical threads (minus node_id). -**M2 — Composite PK + reconcile.** Schema migration; `knox reconcile`; `entries` -fully derived; dedup re-expressed on HCL. Verify: reconcile is idempotent; a DB -with only the log reconstructs `entries`/threads bit-identical to the original. +**M2 — Composite PK + reconcile.** DONE. HCL backfilled from local rowid; +`(node_id, hcl)` locator UNIQUE index; threads carry `cluster_key` (partial +UNIQUE index) making auto-creation idempotent; `knox reconcile` rebuilds +`entries` from the log and re-links threads. Verify: reconcile is idempotent (0 +create/0 link on re-run); a log-only DB reconstructs `entries` and thread +`cluster_key`s bit-identical to the original. **M3 — Peer protocol.** `peers` settings table, `/v1/ping`, `/v1/log` pull, `/v1/obs/batch` push, handshake + periodic anti-entropy, echo suppression. diff --git a/internal/cmd/reconcile.go b/internal/cmd/reconcile.go new file mode 100644 index 0000000..82910b9 --- /dev/null +++ b/internal/cmd/reconcile.go @@ -0,0 +1,63 @@ +package cmd + +import ( + "fmt" + + "github.com/spf13/cobra" + "github.com/david/knox/internal/db" + "github.com/david/knox/internal/watch" +) + +// NewReconcileCmd rebuilds all derived state (entries cache, threads) from the +// append-only observation log. With --dry-run it reports what would change +// without writing. +func NewReconcileCmd(kdb *db.KnoxDB) *cobra.Command { + var dryRun bool + cmd := &cobra.Command{ + Use: "reconcile", + Short: "Rebuild entries and threads from the observation log", + Long: `Reconcile makes derived state converge on the append-only observation log. + +Entries are dropped and rebuilt via deterministic SQL aggregation; the +auto-threader then re-links/create threads idempotently by cluster_key. After a +gossip pull, reconcile brings a node's materialized view in line with whatever +observations it now holds.`, + RunE: func(c *cobra.Command, args []string) error { + if dryRun { + return dryRunReconcile(kdb) + } + return reconcile(kdb) + }, + } + cmd.Flags().BoolVar(&dryRun, "dry-run", false, "report drift without writing") + return cmd +} + +func reconcile(kdb *db.KnoxDB) error { + before, after, err := kdb.RebuildEntriesFromObservations() + if err != nil { + return fmt.Errorf("rebuild entries: %w", err) + } + threader := watch.NewAutoThreader(kdb) + created, linked, err := threader.AutoThread() + if err != nil { + return fmt.Errorf("auto-thread: %w", err) + } + fmt.Printf("reconciled: entries %d -> %d, %d threads created, %d observations linked\n", before, after, created, linked) + return nil +} + +func dryRunReconcile(kdb *db.KnoxDB) error { + // Ground truth from the log (computed in a throwaway way via a count of + // what rebuild would produce) vs the current materialized cache. + current, _ := kdb.EntryCount() + logCount := kdb.ObservationEntryEstimate() + + threader := watch.NewAutoThreader(kdb) + threader.DryRun = true + created, linked, _ := threader.AutoThread() + + fmt.Printf("drift: %d entries current, %d from log (%+d), %d threads would be created, %d obs would be linked\n", + current, logCount, logCount-current, created, linked) + return nil +} \ No newline at end of file diff --git a/internal/db/db.go b/internal/db/db.go index 1b12d26..4cc09b5 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -109,6 +109,7 @@ func Open(path string) (*KnoxDB, error) { "ALTER TABLE observations ADD COLUMN node_id TEXT NOT NULL DEFAULT ''", "ALTER TABLE observations ADD COLUMN hcl INTEGER", "ALTER TABLE observations ADD COLUMN received_at TEXT", + "ALTER TABLE threads ADD COLUMN cluster_key TEXT", } for _, m := range migrations { if _, err := db.Exec(m); err != nil && !strings.Contains(err.Error(), "duplicate column") { @@ -126,10 +127,20 @@ func Open(path string) (*KnoxDB, error) { } kdb.nodeID = nodeID - // Backfill: rows added before node_id existed belong to this node. + // Backfill: rows added before node_id existed belong to this node, and + // their HCL is their local rowid (monotonic, so ordering is preserved). if _, err := db.Exec("UPDATE observations SET node_id=? WHERE node_id=''", nodeID); err != nil { return nil, fmt.Errorf("backfill node_id: %w", err) } + if _, err := db.Exec("UPDATE observations SET hcl=id WHERE hcl IS NULL"); err != nil { + return nil, fmt.Errorf("backfill hcl: %w", err) + } + + // Locator uniqueness: (node_id, hcl) is the merge key for gossip; a given + // node's HCL is strictly monotonic so this never throws a false conflict. + if _, err := db.Exec(`CREATE UNIQUE INDEX IF NOT EXISTS idx_obs_locator ON observations(node_id, hcl)`); err != nil { + return nil, fmt.Errorf("create locator index: %w", err) + } return kdb, nil } @@ -166,6 +177,23 @@ func (k *KnoxDB) NodeID() string { return k.nodeID } func (k *KnoxDB) Close() error { return k.db.Close() } +// EntryCount returns the number of entries in the materialized cache. +func (k *KnoxDB) EntryCount() (int, error) { + var n int + err := k.db.QueryRow("SELECT COUNT(*) FROM entries").Scan(&n) + return n, err +} + +// ObservationEntryEstimate is the number of distinct fingerprints in the +// observation log — the ground-truth size reconcile's rebuild would produce. +func (k *KnoxDB) ObservationEntryEstimate() int { + var n int + if err := k.db.QueryRow("SELECT COUNT(DISTINCT fingerprint) FROM observations").Scan(&n); err != nil { + return -1 + } + return n +} + // RecordObservation appends an immutable observation and updates the entries cache. // Idempotent: if the latest observation for this fingerprint has an identical // content signature, nothing is recorded — re-ingesting unchanged content is a no-op. @@ -379,15 +407,20 @@ func coalesceCreatedAt(s string) string { return time.Now().UTC().Format(time.RFC3339) } -func (k *KnoxDB) RebuildEntriesFromObservations() error { +func (k *KnoxDB) RebuildEntriesFromObservations() (before, after int, err error) { tx, err := k.db.Begin() if err != nil { - return err + return 0, 0, err } defer tx.Rollback() + var b int + if err := tx.QueryRow("SELECT COUNT(*) FROM entries").Scan(&b); err != nil { + return 0, 0, err + } + if _, err := tx.Exec("DELETE FROM entries"); err != nil { - return err + return 0, 0, err } _, err = tx.Exec( `INSERT INTO entries (fingerprint, source_id, source_path, project, content_type, title, summary, first_seen, last_seen, created_at, ref_count, last_confidence) @@ -408,9 +441,16 @@ func (k *KnoxDB) RebuildEntriesFromObservations() error { GROUP BY o.fingerprint`, ) if err != nil { - return fmt.Errorf("rebuild entries: %w", err) + return 0, 0, fmt.Errorf("rebuild entries: %w", err) } - return tx.Commit() + var a int + if err := tx.QueryRow("SELECT COUNT(*) FROM entries").Scan(&a); err != nil { + return 0, 0, err + } + if err := tx.Commit(); err != nil { + return 0, 0, err + } + return b, a, nil } // EntriesByProject returns entries from the materialized cache. @@ -915,15 +955,43 @@ type Thread struct { func ThreadFP(id int64) string { return fmt.Sprintf("thread:%d", id) } +// CreateThread inserts a thread. clusterKey, when non-empty, provides +// idempotency: if a thread with that cluster key already exists no new row is +// created. The existing id is returned with created=false. func (k *KnoxDB) CreateThread(title, motivation, priority, tags, provenance string) (int64, error) { + id, _, err := k.CreateThreadCluster(title, motivation, priority, tags, provenance, "") + return id, err +} + +func (k *KnoxDB) CreateThreadCluster(title, motivation, priority, tags, provenance, clusterKey string) (int64, bool, error) { + if clusterKey != "" { + if existing := k.ThreadByClusterKey(clusterKey); existing != 0 { + return existing, false, nil + } + } res, err := k.db.Exec( - `INSERT INTO threads (title, motivation, priority, tags, provenance) VALUES (?, ?, ?, ?, ?)`, - title, motivation, priority, tags, provenance, + `INSERT INTO threads (title, motivation, priority, tags, provenance, cluster_key) VALUES (?, ?, ?, ?, ?, ?)`, + title, motivation, priority, tags, provenance, clusterKey, ) if err != nil { - return 0, err + return 0, false, err } - return res.LastInsertId() + id, _ := res.LastInsertId() + return id, true, nil +} + +// ThreadByClusterKey returns the id of the thread auto-created for a given +// cluster, or 0 when none exists. +func (k *KnoxDB) ThreadByClusterKey(clusterKey string) int64 { + if clusterKey == "" { + return 0 + } + var id int64 + err := k.db.QueryRow(`SELECT id FROM threads WHERE cluster_key=?`, clusterKey).Scan(&id) + if err != nil { + return 0 + } + return id } func (k *KnoxDB) ListThreads(status string) ([]Thread, error) { @@ -1043,7 +1111,7 @@ func (k *KnoxDB) AutoLinkThreadObservations(threadID int64, fingerprints []strin } var obsID int64 if err := tx.QueryRow( - `SELECT id FROM observations WHERE fingerprint=? ORDER BY id DESC LIMIT 1`, fp, + `SELECT id FROM observations WHERE fingerprint=? ORDER BY hcl DESC, id DESC LIMIT 1`, fp, ).Scan(&obsID); err != nil { continue } diff --git a/internal/db/schema.go b/internal/db/schema.go index 6a37d02..278dd43 100644 --- a/internal/db/schema.go +++ b/internal/db/schema.go @@ -101,9 +101,12 @@ CREATE TABLE IF NOT EXISTS threads ( updated_at TEXT DEFAULT (datetime('now')), resolved_at TEXT, tags TEXT DEFAULT '[]', - provenance TEXT DEFAULT '{}' + provenance TEXT DEFAULT '{}', + cluster_key TEXT ); +CREATE UNIQUE INDEX IF NOT EXISTS idx_threads_cluster ON threads(cluster_key) WHERE cluster_key IS NOT NULL AND cluster_key != ''; + -- SETTINGS: key-value store for runtime configuration CREATE TABLE IF NOT EXISTS settings ( key TEXT PRIMARY KEY, diff --git a/internal/watch/threader.go b/internal/watch/threader.go index a0cd752..ee2d7d2 100644 --- a/internal/watch/threader.go +++ b/internal/watch/threader.go @@ -1,10 +1,12 @@ package watch import ( + "crypto/sha256" "encoding/json" "fmt" "log" "os" + "sort" "strconv" "strings" "time" @@ -80,8 +82,14 @@ func (t *AutoThreader) AutoThread() (created, linked int, err error) { continue } - // Fold into an existing active thread if one already covers these terms. - if existing := t.findExisting(c); existing > 0 { + // Fold into the exact thread this cluster keyed to (idempotent across + // nodes/reconciles); otherwise fall back to the keyword heuristic which + // also matches human-edited threads that keep their auto cluster key. + existing := t.DB.ThreadByClusterKey(clusterKey(c)) + if existing == 0 { + existing = t.findExisting(c) + } + if existing > 0 { if t.DryRun { fmt.Printf("[threader] (dry-run) would link %d obs into existing thread #%d: %s\n", len(fps), existing, c.Name) continue @@ -118,7 +126,7 @@ func (t *AutoThreader) AutoThread() (created, linked int, err error) { continue } - id, err := t.DB.CreateThread(title, motivation, priority, tags, string(provJSON)) + id, isNew, err := t.DB.CreateThreadCluster(title, motivation, priority, tags, string(provJSON), clusterKey(c)) if err != nil { log.Printf("[knox] threader create err: %v", err) continue @@ -126,12 +134,34 @@ func (t *AutoThreader) AutoThread() (created, linked int, err error) { if n, err := t.DB.AutoLinkThreadObservations(id, fps); err == nil { linked += n } - created++ - log.Printf("[knox] auto-thread #%d: %s (%d obs, %s)", id, title, len(fps), priority) + if isNew { + created++ + } + log.Printf("[knox] auto-thread %s #%d: %s (%d obs, %s)", statusWord(isNew), id, title, len(fps), priority) } return created, linked, nil } +func statusWord(isNew bool) string { + if isNew { + return "created" + } + return "extended" +} + +// clusterKey returns a deterministic content hash keying a thread to the topic +// cluster that generated it. Sorted keywords make it machine-independent, so +// two nodes auto-creating the same cluster converge on the same key (idempotent +// thread creation via CreateThreadCluster). +func clusterKey(c index.TopicCluster) string { + kws := make([]string, len(c.Keywords)) + copy(kws, c.Keywords) + sort.Strings(kws) + joined := "thread:cluster:" + strings.Join(kws, "\x00") + sum := sha256.Sum256([]byte(joined)) + return fmt.Sprintf("cluster:%x", sum[:16]) +} + // crossesBar decides whether a cluster reflects real intent. func (t *AutoThreader) crossesBar(c index.TopicCluster, fps []string, now time.Time) bool { if len(c.Entries) < t.MinEntries { diff --git a/main.go b/main.go index 6eed36a..a1b1228 100644 --- a/main.go +++ b/main.go @@ -63,6 +63,7 @@ and maintains a searchable index. Use 'knox watch' for daemon mode.`, root.AddCommand(knoxcmd.NewTopicsCmd(kdb)) root.AddCommand(knoxcmd.NewGiteaCmd(kdb)) root.AddCommand(knoxcmd.NewGitCmd(kdb)) + root.AddCommand(knoxcmd.NewReconcileCmd(kdb)) root.AddCommand(newServeCmd(kdb)) // MCP server subcommand