feat: M2 composite locator + idle reconcile

Refs #1

- observations: hcl backfilled from local rowid; (node_id, hcl) UNIQUE
  locator index becomes the gossip merge key
- threads: cluster_key column + partial UNIQUE index; CreateThreadCluster
  is idempotent, threader folds into exact cluster_key before heuristic
- AutoLinkThreadObservations dedup re-expressed on hcl DESC
- new `knox reconcile [--dry-run]` rebuilds entries from the observation
  log and re-links threads idempotently (entry count, thread cluster_key
  verified bit-identical from log-only DB)

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.

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
+}

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
 		}

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,

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
 		}
+		if isNew {
 			created++
-		log.Printf("[knox] auto-thread #%d: %s (%d obs, %s)", id, title, len(fps), priority)
+		}
+		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 {

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