fix: harden gossip, HLC restarts, watcher races, MCP args, pagination (#3)

Implements the top findings from the codebase review, verified with tests and live CLI/MCP checks.

**Gossip integrity**
- Push validation: 4 MiB body cap, 1000-row batch cap; rows claiming the local node id (vector-poisoning), empty node ids, and negative HCLs rejected (internal/watch/gossip.go, internal/db/gossip.go)
- Reconcile-on-pull: Run returns the pulled count, syncGossip rebuilds derived state when > 0 — entry-count comparison could never fire, so synced observations never materialized into searchable entries

**Data-layer safety**
- HLC resumed from MAX(hcl) at Open (hlc.SeekTo): a restart with a regressed wall clock cannot reissue values the (node_id, hcl) locator and pull cursors depend on
- Writer serialization: _txlock=immediate DSN + SetMaxOpenConns(1) + per-KnoxDB mutex around RecordObservation's check-then-insert dedup (closes duplicate-row race)

**Watch daemon**
- Ticker guard flags now atomic.Bool (was a cross-goroutine data race)
- Trailing-edge per-path debounce (timer-based, pruned on fire/delete)
- Recursive watches (startup tree walk + watcher.Add on dir Create); Rename re-ingests, Remove cancels pending ingests

**MCP + CLI**
- Strict arg validation, no silent clamping: thread_id 0 errors instead of renaming thread #1; empty knox_thread_link {} errors instead of false success; thread existence checked before writes; golden-thread tool nil-safe
- --page 0 errors instead of panicking; query/recent pagination actually pages (page x limit)

**Tests** (new internal/hlc and internal/db packages): SeekTo monotonicity, concurrent dedup race, push validation, reopen HCL monotonicity, batch caps, self-spoof rejection, idempotency on observation counts.

Verified: go build, go vet, full suite with -race, live MCP stdio transcripts against a scratch DB.
Reviewed-on: #3
Co-authored-by: David Gwilliam <dhgwilliam@gmail.com>
Co-committed-by: David Gwilliam <dhgwilliam@gmail.com>

docs/gossip-protocol.md

@@ -125,14 +125,16 @@ a single node the rowid remains monotonic, so the current dedup query
 
 ### 5.1 Transport
 
-Plain HTTP/JSON on a per-node advertized address (default port `8931`). Nodes
-discover peers via a static list in `settings` (M3). mDNS/rendezvous is future
-work.
+Plain HTTP/JSON on a per-node advertized address (default port `8931`). Peers
+are seeded from a static list (`KNOX_PEERS`), then the swarm discovers itself:
+each node advertises its known peer addresses in `/v1/ping`, and every sweep
+enqueues newly-learned nodes for direct contact (membership gossip — no relay of
+observations). mDNS/rendezvous is future work.
 
 Endpoints:
 
 ```
-GET  /v1/ping                       → { node_id, name, max_hcl }
+GET  /v1/ping                       → { node_id, name, max_hcl, peers: [{peer_id, addr, name}] }
 GET  /v1/log?after=<seq>&node=<id>  → { cursor, rows: [observation...] }   # pull
 POST /v1/obs/batch                  → body: [observation...]; reply: { accepted n, conflict n }  # push
 GET  /v1/diff                       → divergence summary (M4)
@@ -142,6 +144,11 @@ GET  /v1/diff                       → divergence summary (M4)
 
 - **Knowledge vector:** each node tracks `peer_id → max_hcl consumed`. Anti-entropy
   is a pull: periodically (and on handshake) query each peer's `/v1/log?after=...`.
+- **Membership gossip:** `/v1/ping` includes the responding node's known peers
+  (`peer_id`, `addr`, `name`). The caller merges them into its `peers` table and
+  enqueues their addresses for direct sweeps. A new node therefore joins the
+  whole swarm by configuring just one seed peer. Membership flows independently
+  of data — a node never relays another's observations, only its address.
 - **Push:** on a new local observation, best-effort `POST /v1/obs/batch` to known
   peers. A node does **not** re-broadcast something it merely received (that peer
   already has it and will pull from its origin) — this is the echo/loop
@@ -242,6 +249,17 @@ reach a booting node. Config via env `KNOX_PEER_ADDR` / `KNOX_PEERS`.
 Verified e2e: two daemons, `diff` previewed 1655 peer-only fingerprints, `sync`
 pulled all and converged B to 1655 observations.
 
+**M5 — Prometheus metrics.** DONE. `/metrics` served on a dedicated port
+(`KNOX_METRICS_ADDR`, default `:8932`, separate from gossip). Uses
+`prometheus/client_golang`; ships with the Go runtime and process collectors.
+Gauge set: DB-derived gauges refreshed per scrape (`observations_total{source_id}`,
+`observations_last_24h`, `entries`, `projects`, `sessions`, `pending_reflections`,
+`threads_total{status}`, `peers`, `observations_by_node{node_id}`,
+`knowledge_max_hcl{node_id}`) plus live gossip counters
+(`pulls/pushes`, `observations_pulled/pushed`, `errors`) incremented during the
+anti-entropy sweep, and `knox_node_info{node_id,name}` for scrape identity
+(tracks issue #2).
+
 ## 9. Future Work (explicitly out of M1–M4)
 
 - mDNS / rendezvous peer discovery.

go.mod

@@ -5,21 +5,30 @@ go 1.26
 require (
 	github.com/fsnotify/fsnotify v1.8.0
 	github.com/mark3labs/mcp-go v0.17.0
+	github.com/prometheus/client_golang v1.24.1
 	github.com/spf13/cobra v1.9.1
 	modernc.org/sqlite v1.37.1
 )
 
 require (
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/ncruces/go-strftime v0.1.9 // indirect
+	github.com/prometheus/client_model v0.6.2 // indirect
+	github.com/prometheus/common v0.70.1 // indirect
+	github.com/prometheus/procfs v0.21.1 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
 	github.com/spf13/pflag v1.0.6 // indirect
 	github.com/yosida95/uritemplate/v3 v3.0.2 // indirect
 	golang.org/x/exp v0.0.0-20250408133849-7e4ce0ab07d0 // indirect
-	golang.org/x/sys v0.33.0 // indirect
+	golang.org/x/sys v0.47.0 // indirect
+	google.golang.org/protobuf v1.36.11 // indirect
 	modernc.org/libc v1.65.7 // indirect
 	modernc.org/mathutil v1.7.1 // indirect
 	modernc.org/memory v1.11.0 // indirect

go.sum

@@ -1,3 +1,7 @@
+github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
+github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -5,20 +9,36 @@ github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkp
 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
 github.com/fsnotify/fsnotify v1.8.0 h1:dAwr6QBTBZIkG8roQaJjGof0pp0EeF+tNV7YBP3F/8M=
 github.com/fsnotify/fsnotify v1.8.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0=
+github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
+github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
 github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs=
 github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/klauspost/compress v1.19.1 h1:VsB4HPswih7mmZ8WleSFQ75c/Ui1M4trX5oAsJnhSlk=
+github.com/klauspost/compress v1.19.1/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
+github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
+github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
 github.com/mark3labs/mcp-go v0.17.0 h1:5Ps6T7qXr7De/2QTqs9h6BKeZ/qdeUeGrgM5lPzi930=
 github.com/mark3labs/mcp-go v0.17.0/go.mod h1:KmJndYv7GIgcPVwEKJjNcbhVQ+hJGJhrCCB/9xITzpE=
 github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
 github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
 github.com/ncruces/go-strftime v0.1.9 h1:bY0MQC28UADQmHmaF5dgpLmImcShSi2kHU9XLdhx/f4=
 github.com/ncruces/go-strftime v0.1.9/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU=
+github.com/prometheus/client_golang v1.24.1/go.mod h1:F+oSRECHg4sse5ucfYpYDeIv/hu68Zo0uoHKetWnzcE=
+github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
+github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
+github.com/prometheus/common v0.70.1 h1:1HvjP4D5oL3t8RsPlwxA9onvvStjtIHYE5XuuwOi/PY=
+github.com/prometheus/common v0.70.1/go.mod h1:VdFUQDMZK3VLkurFUVhia6uys/0suUp86TJz5qbJRhc=
+github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+OJI=
+github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
@@ -26,21 +46,27 @@ github.com/spf13/cobra v1.9.1 h1:CXSaggrXdbHK9CF+8ywj8Amf7PBRmPCOJugH954Nnlo=
 github.com/spf13/cobra v1.9.1/go.mod h1:nDyEzZ8ogv936Cinf6g1RU9MRY64Ir93oCnqb9wxYW0=
 github.com/spf13/pflag v1.0.6 h1:jFzHGLGAlb3ruxLB8MhbI6A8+AQX/2eW4qeyNZXNp2o=
 github.com/spf13/pflag v1.0.6/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
-github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
-github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4=
 github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
+go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
+go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
 golang.org/x/exp v0.0.0-20250408133849-7e4ce0ab07d0 h1:R84qjqJb5nVJMxqWYb3np9L5ZsaDtB+a39EqjV0JSUM=
 golang.org/x/exp v0.0.0-20250408133849-7e4ce0ab07d0/go.mod h1:S9Xr4PYopiDyqSyp5NjCrhFrqg6A5zA2E/iPHPhqnS8=
 golang.org/x/mod v0.24.0 h1:ZfthKaKaT4NrhGVZHO1/WDTwGES4De8KtWO0SIbNJMU=
 golang.org/x/mod v0.24.0/go.mod h1:IXM97Txy2VM4PJ3gI61r1YEk/gAj6zAHN3AdZt6S9Ww=
-golang.org/x/sync v0.14.0 h1:woo0S4Yywslg6hp4eUFjTVOyKt0RookbpAHG4c1HmhQ=
-golang.org/x/sync v0.14.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
+golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM=
+golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw=
-golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
+golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
+golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/tools v0.33.0 h1:4qz2S3zmRxbGIhDIAgjxvFutSvH5EfnsYrRBj0UI0bc=
 golang.org/x/tools v0.33.0/go.mod h1:CIJMaWEY88juyUfo7UbgPqbC8rU2OqfAV1h2Qp0oMYI=
+google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
+google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

internal/cmd/browser.go

@@ -5,9 +5,9 @@ import (
 	"log"
 	"strings"
 
-	"github.com/spf13/cobra"
 	"github.com/david/knox/internal/db"
 	"github.com/david/knox/internal/ingest"
+	"github.com/spf13/cobra"
 )
 
 func NewBrowserCmd(kdb *db.KnoxDB) *cobra.Command {
@@ -65,10 +65,10 @@ Run periodically to capture browsing exhaust.`,
 			for _, r := range results {
 				domains[extractDomain(r.Summary)]++
 			}
-	fmt.Println("\nTop domains:")
-	for _, d := range topDomains(domains, 10) {
-		fmt.Printf("  %-40s %d\n", d.Name, d.Count)
-	}
+			fmt.Println("\nTop domains:")
+			for _, d := range topDomains(domains, 10) {
+				fmt.Printf("  %-40s %d\n", d.Name, d.Count)
+			}
 
 			return nil
 		},

internal/cmd/gitea.go

@@ -4,9 +4,9 @@ import (
 	"fmt"
 	"log"
 
-	"github.com/spf13/cobra"
 	"github.com/david/knox/internal/db"
 	"github.com/david/knox/internal/ingest"
+	"github.com/spf13/cobra"
 )
 
 func NewGiteaCmd(kdb *db.KnoxDB) *cobra.Command {

internal/cmd/gossip.go

@@ -3,9 +3,9 @@ package cmd
 import (
 	"fmt"
 
-	"github.com/spf13/cobra"
 	"github.com/david/knox/internal/db"
 	"github.com/david/knox/internal/watch"
+	"github.com/spf13/cobra"
 )
 
 // NewGossipCmd exposes peer status/diff for the gossip protocol.
@@ -61,7 +61,7 @@ func NewGossipCmd(kdb *db.KnoxDB) *cobra.Command {
 				if len(peers) == 0 {
 					return fmt.Errorf("no peers configured (set KNOX_PEERS)")
 				}
-				watch.Run(kdb, peers)
+				watch.Run(kdb, nil, peers)
 				created, linked, err := watch.Reconcile(kdb)
 				if err != nil {
 					return err
@@ -154,4 +154,4 @@ func min(a, b int) int {
 		return a
 	}
 	return b
-}
+}

internal/cmd/ingest.go

@@ -5,10 +5,10 @@ import (
 	"log"
 	"path/filepath"
 
-	"github.com/spf13/cobra"
 	"github.com/david/knox/internal/db"
 	"github.com/david/knox/internal/ingest"
 	"github.com/david/knox/internal/watch"
+	"github.com/spf13/cobra"
 )
 
 func NewIngestCmd(kdb *db.KnoxDB) *cobra.Command {
@@ -27,92 +27,92 @@ func NewIngestCmd(kdb *db.KnoxDB) *cobra.Command {
 				ingest.NewSkillsIngester(),
 			}
 
-		var totalNew, totalUpdated int
+			var totalNew, totalUpdated int
 
-		record := func(result *ingest.IngestResult) {
-			_, isNew, err := kdb.RecordObservation(db.ObservationRecord{
-				Fingerprint:     result.Fingerprint,
-				SourceID:        result.SourceID,
-				SourcePath:      result.SourcePath,
-				Project:         result.Project,
-				ContentType:     result.ContentType,
-				Title:           result.Title,
-				Summary:         result.Summary,
-				CreatedAt:       result.CreatedAt,
-				LineStart:       result.LineStart,
-				LineEnd:         result.LineEnd,
-				Confidence:      result.Confidence,
-				IngesterVersion: result.IngesterVersion,
-				Trigger:         "ingest",
-				Provenance:      ingest.ProvenanceJSON(result.Provenance),
-			})
-			if err != nil {
-				log.Printf("[knox] db error: %v", err)
-				return
-			}
-			if isNew {
-				totalNew++
-			} else {
-				totalUpdated++
-			}
-
-			if result.SourceID == "opencode-session" {
-				sessionID, _ := result.Provenance["session_id"].(string)
-				if sessionID != "" {
-					kdb.UpsertSession(sessionID, result.Project, result.Title, "active")
+			record := func(result *ingest.IngestResult) {
+				_, isNew, err := kdb.RecordObservation(db.ObservationRecord{
+					Fingerprint:     result.Fingerprint,
+					SourceID:        result.SourceID,
+					SourcePath:      result.SourcePath,
+					Project:         result.Project,
+					ContentType:     result.ContentType,
+					Title:           result.Title,
+					Summary:         result.Summary,
+					CreatedAt:       result.CreatedAt,
+					LineStart:       result.LineStart,
+					LineEnd:         result.LineEnd,
+					Confidence:      result.Confidence,
+					IngesterVersion: result.IngesterVersion,
+					Trigger:         "ingest",
+					Provenance:      ingest.ProvenanceJSON(result.Provenance),
+				})
+				if err != nil {
+					log.Printf("[knox] db error: %v", err)
+					return
 				}
-			}
-		}
-
-		for _, ing := range ingesters {
-			sid := ing.SourceID()
-			log.Printf("[knox] ingesting %s...", sid)
-
-			for _, dir := range dirs {
-				patterns := []string{
-					dir + "/*",
-					dir + "/*/SKILL.md",
+				if isNew {
+					totalNew++
+				} else {
+					totalUpdated++
 				}
-				for _, pattern := range patterns {
-					entries, _ := filepath.Glob(pattern)
-					for _, path := range entries {
-						if !watch.MatchesIngester(path, sid) {
-							continue
-						}
 
-						result, err := ing.Ingest(path)
-						if err != nil || result == nil {
-							continue
-						}
-						record(result)
+				if result.SourceID == "opencode-session" {
+					sessionID, _ := result.Provenance["session_id"].(string)
+					if sessionID != "" {
+						kdb.UpsertSession(sessionID, result.Project, result.Title, "active")
 					}
 				}
 			}
-		}
 
-		// Non-file sources: obsidian vault, browser history, gitea
-		if vault, err := ingest.DetectObsidianVault(); err == nil {
-			log.Printf("[knox] ingesting obsidian...")
-			if results, err := ingest.NewObsidianIngester(vault).IngestAll(); err == nil {
+			for _, ing := range ingesters {
+				sid := ing.SourceID()
+				log.Printf("[knox] ingesting %s...", sid)
+
+				for _, dir := range dirs {
+					patterns := []string{
+						dir + "/*",
+						dir + "/*/SKILL.md",
+					}
+					for _, pattern := range patterns {
+						entries, _ := filepath.Glob(pattern)
+						for _, path := range entries {
+							if !watch.MatchesIngester(path, sid) {
+								continue
+							}
+
+							result, err := ing.Ingest(path)
+							if err != nil || result == nil {
+								continue
+							}
+							record(result)
+						}
+					}
+				}
+			}
+
+			// Non-file sources: obsidian vault, browser history, gitea
+			if vault, err := ingest.DetectObsidianVault(); err == nil {
+				log.Printf("[knox] ingesting obsidian...")
+				if results, err := ingest.NewObsidianIngester(vault).IngestAll(); err == nil {
+					for _, r := range results {
+						record(r)
+					}
+				}
+			}
+			log.Printf("[knox] ingesting browser-history...")
+			if results, err := ingest.NewBrowserHistoryIngester().IngestAll(); err == nil {
 				for _, r := range results {
 					record(r)
 				}
 			}
-		}
-		log.Printf("[knox] ingesting browser-history...")
-		if results, err := ingest.NewBrowserHistoryIngester().IngestAll(); err == nil {
-			for _, r := range results {
-				record(r)
+			log.Printf("[knox] ingesting gitea...")
+			if results, err := ingest.NewGiteaIngester().IngestAll(); err == nil {
+				for _, r := range results {
+					record(r)
+				}
 			}
-		}
-		log.Printf("[knox] ingesting gitea...")
-		if results, err := ingest.NewGiteaIngester().IngestAll(); err == nil {
-			for _, r := range results {
-				record(r)
-			}
-		}
 
-		fmt.Printf("\nIngest complete: %d new entries, %d updated\n", totalNew, totalUpdated)
+			fmt.Printf("\nIngest complete: %d new entries, %d updated\n", totalNew, totalUpdated)
 		},
 	}
 	cmd.Flags().StringSliceVarP(&dirs, "dir", "d", nil, "Directories to scan (default: opencode storage/log + skills)")

internal/cmd/mcp.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"math"
 	"strings"
 
 	"github.com/david/knox/internal/db"
@@ -50,8 +51,14 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer {
 	)
 
 	s.AddTool(searchTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
-		query, _ := req.Params.Arguments["query"].(string)
-		limit := clampInt(req, "limit", 10, 1, 50)
+		query, err := requiredStringArg(req, "query")
+		if err != nil {
+			return errorResult(err.Error()), nil
+		}
+		limit, err := optionalIntArg(req, "limit", 10, 1, 50)
+		if err != nil {
+			return errorResult(err.Error()), nil
+		}
 		detail := parseDetail(getString(req, "detail", "normal"))
 		scope := getString(req, "scope", "all")
 
@@ -76,7 +83,10 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer {
 	)
 
 	s.AddTool(recentTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
-		limit := clampInt(req, "limit", 10, 1, 50)
+		limit, err := optionalIntArg(req, "limit", 10, 1, 50)
+		if err != nil {
+			return errorResult(err.Error()), nil
+		}
 		detail := parseDetail(getString(req, "detail", "normal"))
 		scope := getString(req, "scope", "all")
 
@@ -99,7 +109,10 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer {
 	)
 
 	s.AddTool(getTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
-		fp := getString(req, "fingerprint", "")
+		fp, err := requiredStringArg(req, "fingerprint")
+		if err != nil {
+			return errorResult(err.Error()), nil
+		}
 
 		entry, err := kdb.FindEntry(fp)
 		if err != nil {
@@ -165,7 +178,10 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer {
 	)
 
 	s.AddTool(countTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
-		query, _ := req.Params.Arguments["query"].(string)
+		query, err := requiredStringArg(req, "query")
+		if err != nil {
+			return errorResult(err.Error()), nil
+		}
 		results, err := kdb.Search(query, 500)
 		if err != nil {
 			return errorResult(err.Error()), nil
@@ -235,7 +251,10 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer {
 	)
 
 	s.AddTool(threadCreateTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
-		title := getString(req, "title", "")
+		title, err := requiredStringArg(req, "title")
+		if err != nil {
+			return errorResult(err.Error()), nil
+		}
 		motivation := getString(req, "motivation", "")
 		priority := getString(req, "priority", "medium")
 		provenance := getString(req, "provenance", "{}")
@@ -257,7 +276,10 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer {
 	)
 
 	s.AddTool(threadUpdateTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
-		id := int64(clampInt(req, "thread_id", 0, 1, 999999))
+		id, err := requiredIntArg(req, "thread_id", 1, 999999)
+		if err != nil {
+			return errorResult(err.Error()), nil
+		}
 		title := getString(req, "title", "")
 		motivation := getString(req, "motivation", "")
 		priority := getString(req, "priority", "")
@@ -281,7 +303,10 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer {
 	)
 
 	s.AddTool(threadDraftTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
-		limit := clampInt(req, "limit", 20, 1, 200)
+		limit, err := optionalIntArg(req, "limit", 20, 1, 200)
+		if err != nil {
+			return errorResult(err.Error()), nil
+		}
 		threads, err := kdb.ListThreads("")
 		if err != nil {
 			return errorResult(err.Error()), nil
@@ -329,9 +354,18 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer {
 	)
 
 	s.AddTool(threadLinkTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
-		threadID := int64(clampInt(req, "thread_id", 0, 1, 999999))
-		obsID := int64(clampInt(req, "observation_id", 0, 1, 999999))
+		threadID, err := requiredIntArg(req, "thread_id", 1, 999999)
+		if err != nil {
+			return errorResult(err.Error()), nil
+		}
+		obsID, err := requiredIntArg(req, "observation_id", 1, 999999)
+		if err != nil {
+			return errorResult(err.Error()), nil
+		}
 		relevance := getString(req, "relevance", "")
+		if err := threadExists(kdb, threadID); err != nil {
+			return errorResult(err.Error()), nil
+		}
 		if err := kdb.LinkObservationToThread(threadID, obsID, relevance); err != nil {
 			return errorResult(err.Error()), nil
 		}
@@ -345,7 +379,10 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer {
 	)
 
 	s.AddTool(threadSearchTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
-		query := getString(req, "query", "")
+		query, err := requiredStringArg(req, "query")
+		if err != nil {
+			return errorResult(err.Error()), nil
+		}
 		threads, err := kdb.SearchThreadsByMotivation(query)
 		if err != nil {
 			return errorResult(err.Error()), nil
@@ -372,9 +409,18 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer {
 	)
 
 	s.AddTool(threadLinkEntryTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
-		threadID := int64(clampInt(req, "thread_id", 0, 1, 999999))
-		fp := getString(req, "fingerprint", "")
+		threadID, err := requiredIntArg(req, "thread_id", 1, 999999)
+		if err != nil {
+			return errorResult(err.Error()), nil
+		}
+		fp, err := requiredStringArg(req, "fingerprint")
+		if err != nil {
+			return errorResult(err.Error()), nil
+		}
 		relation := getString(req, "relation", "produced")
+		if err := threadExists(kdb, threadID); err != nil {
+			return errorResult(err.Error()), nil
+		}
 		if err := kdb.LinkEntryToThread(threadID, fp, relation); err != nil {
 			return errorResult(err.Error()), nil
 		}
@@ -390,9 +436,21 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer {
 	)
 
 	s.AddTool(threadRelateTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
-		parentID := int64(clampInt(req, "parent_id", 0, 1, 999999))
-		childID := int64(clampInt(req, "child_id", 0, 1, 999999))
+		parentID, err := requiredIntArg(req, "parent_id", 1, 999999)
+		if err != nil {
+			return errorResult(err.Error()), nil
+		}
+		childID, err := requiredIntArg(req, "child_id", 1, 999999)
+		if err != nil {
+			return errorResult(err.Error()), nil
+		}
 		relation := getString(req, "relation", "spawned")
+		if err := threadExists(kdb, parentID); err != nil {
+			return errorResult(err.Error()), nil
+		}
+		if err := threadExists(kdb, childID); err != nil {
+			return errorResult(err.Error()), nil
+		}
 		if err := kdb.LinkEntryToThread(childID, db.ThreadFP(parentID), relation); err != nil {
 			return errorResult(err.Error()), nil
 		}
@@ -406,29 +464,67 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer {
 	)
 
 	s.AddTool(goldenTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
-		threadID := int64(clampInt(req, "thread_id", 0, 0, 999999))
-
-		if threadID == 0 {
-			// Query or clear
-			current, _ := kdb.GoldenThreadID()
+		// Absent thread_id: query the current golden thread.
+		if _, ok := req.Params.Arguments["thread_id"]; !ok {
+			current, err := kdb.GoldenThreadID()
+			if err != nil {
+				return errorResult(err.Error()), nil
+			}
 			if current == 0 {
 				return mcp.NewToolResultText("No golden thread set."), nil
 			}
-			// If thread_id was explicitly 0 and a golden exists, clear it
-			if _, ok := req.Params.Arguments["thread_id"]; ok {
-				kdb.SetGoldenThread(0)
-				t, _ := kdb.GetThread(current)
-				return mcp.NewToolResultText(fmt.Sprintf("Golden thread cleared (was #%d: %s)", current, t.Title)), nil
+			t, err := kdb.GetThread(current)
+			if err != nil {
+				return errorResult(err.Error()), nil
+			}
+			if t == nil {
+				return errorResult(fmt.Sprintf("Golden thread #%d no longer exists", current)), nil
 			}
-			t, _ := kdb.GetThread(current)
 			return mcp.NewToolResultText(fmt.Sprintf("Golden thread: #%d %s [%s]\n  %s", t.ID, t.Title, t.Status, t.Motivation)), nil
 		}
 
+		threadID, err := requiredIntArg(req, "thread_id", 0, 999999)
+		if err != nil {
+			return errorResult(err.Error()), nil
+		}
+		if threadID == 0 {
+			// Explicit 0 clears the golden thread.
+			current, err := kdb.GoldenThreadID()
+			if err != nil {
+				return errorResult(err.Error()), nil
+			}
+			if current == 0 {
+				return mcp.NewToolResultText("No golden thread set."), nil
+			}
+			if err := kdb.SetGoldenThread(0); err != nil {
+				return errorResult(err.Error()), nil
+			}
+			t, err := kdb.GetThread(current)
+			if err != nil {
+				return errorResult(err.Error()), nil
+			}
+			name := "?"
+			if t != nil {
+				name = t.Title
+			}
+			return mcp.NewToolResultText(fmt.Sprintf("Golden thread cleared (was #%d: %s)", current, name)), nil
+		}
+
+		if err := threadExists(kdb, threadID); err != nil {
+			return errorResult(err.Error()), nil
+		}
 		if err := kdb.SetGoldenThread(threadID); err != nil {
 			return errorResult(err.Error()), nil
 		}
-		t, _ := kdb.GetThread(threadID)
-		return mcp.NewToolResultText(fmt.Sprintf("Golden thread set to #%d: %s", threadID, t.Title)), nil
+		t, err := kdb.GetThread(threadID)
+		if err != nil {
+			return errorResult(err.Error()), nil
+		}
+		name := "?"
+		if t != nil {
+			name = t.Title
+		}
+		return mcp.NewToolResultText(fmt.Sprintf("Golden thread set to #%d: %s", threadID, name)), nil
 	})
 
 	// ─── knox_topics ───────────────────────────────────────────
@@ -438,7 +534,10 @@ func NewMCPServer(kdb *db.KnoxDB) *server.MCPServer {
 	)
 
 	s.AddTool(topicsTool, func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
-		limit := clampInt(req, "limit", 10, 1, 30)
+		limit, err := optionalIntArg(req, "limit", 10, 1, 30)
+		if err != nil {
+			return errorResult(err.Error()), nil
+		}
 		entries, _ := kdb.RecentEntries(2000)
 		tfidf := index.BuildTFIDF(entries)
 		clusters := tfidf.Cluster(2, limit)
@@ -618,18 +717,59 @@ func getString(req mcp.CallToolRequest, key, def string) string {
 	return def
 }
 
-func clampInt(req mcp.CallToolRequest, key string, def, min, max int) int {
-	if v, ok := req.Params.Arguments[key].(float64); ok {
-		n := int(v)
-		if n < min {
-			return min
-		}
-		if n > max {
-			return max
-		}
-		return n
+// requiredStringArg returns a non-empty string argument or a descriptive error.
+// Empty/missing/wrong-type values are rejected rather than silently defaulted:
+// LLM clients routinely omit or zero value params, and a silent default writes
+// to the wrong thread or reports false success.
+func requiredStringArg(req mcp.CallToolRequest, key string) (string, error) {
+	v, _ := req.Params.Arguments[key].(string)
+	v = strings.TrimSpace(v)
+	if v == "" {
+		return "", fmt.Errorf("%s is required and must be a non-empty string", key)
 	}
-	return def
+	return v, nil
+}
+
+// requiredIntArg returns an integer argument validated against [min, max].
+func requiredIntArg(req mcp.CallToolRequest, key string, min, max int64) (int64, error) {
+	v, ok := req.Params.Arguments[key].(float64)
+	if !ok || v != math.Trunc(v) {
+		return 0, fmt.Errorf("%s is required and must be an integer", key)
+	}
+	n := int64(v)
+	if n < min || n > max {
+		return 0, fmt.Errorf("%s must be in [%d..%d], got %d", key, min, max, n)
+	}
+	return n, nil
+}
+
+// optionalIntArg validates a present numeric argument, defaulting when absent.
+func optionalIntArg(req mcp.CallToolRequest, key string, def, min, max int) (int, error) {
+	v, ok := req.Params.Arguments[key].(float64)
+	if !ok {
+		return def, nil
+	}
+	if v != math.Trunc(v) {
+		return 0, fmt.Errorf("%s must be an integer", key)
+	}
+	n := int(v)
+	if n < min || n > max {
+		return 0, fmt.Errorf("%s must be in [%d..%d], got %d", key, min, max, n)
+	}
+	return n, nil
+}
+
+// threadExists is a guard for write tools: link/relate/set-golden must not
+// silently accept ids with no matching thread.
+func threadExists(kdb *db.KnoxDB, id int64) error {
+	t, err := kdb.GetThread(id)
+	if err != nil {
+		return err
+	}
+	if t == nil {
+		return fmt.Errorf("thread #%d not found", id)
+	}
+	return nil
 }
 
 func shortFP(fp string) string {

internal/cmd/obsidian.go

@@ -4,9 +4,9 @@ import (
 	"fmt"
 	"log"
 
-	"github.com/spf13/cobra"
 	"github.com/david/knox/internal/db"
 	"github.com/david/knox/internal/ingest"
+	"github.com/spf13/cobra"
 )
 
 func NewObsidianCmd(kdb *db.KnoxDB) *cobra.Command {

internal/cmd/reconcile.go

@@ -3,9 +3,9 @@ package cmd
 import (
 	"fmt"
 
-	"github.com/spf13/cobra"
 	"github.com/david/knox/internal/db"
 	"github.com/david/knox/internal/watch"
+	"github.com/spf13/cobra"
 )
 
 // NewReconcileCmd rebuilds all derived state (entries cache, threads) from the
@@ -51,4 +51,4 @@ func dryRunReconcile(kdb *db.KnoxDB) error {
 	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/cmd/status.go

@@ -6,14 +6,14 @@ import (
 	"os"
 	"strings"
 
-	"github.com/spf13/cobra"
 	"github.com/david/knox/internal/db"
 	"github.com/david/knox/internal/index"
+	"github.com/spf13/cobra"
 )
 
 func paginate(entries []db.Entry, page, limit int) []db.Entry {
 	start := (page - 1) * limit
-	if start >= len(entries) {
+	if start < 0 || start >= len(entries) {
 		return nil
 	}
 	end := start + limit
@@ -74,16 +74,22 @@ func NewQueryCmd(kdb *db.KnoxDB) *cobra.Command {
 			if query == "" {
 				return fmt.Errorf("search query required (positional arg or stdin pipe)")
 			}
-			results, err := kdb.Search(query, limit)
+			if page < 1 || limit < 1 {
+				return fmt.Errorf("page and limit must be >= 1 (got page=%d limit=%d)", page, limit)
+			}
+			// Fetch enough rows to cover the requested page: Search caps at its
+			// limit argument, so slicing a limit-sized result set could never
+			// reach page 2+.
+			all, err := kdb.Search(query, page*limit)
 			if err != nil {
 				return err
 			}
-			results = paginate(results, page, limit)
+			results := paginate(all, page, limit)
 			if len(results) == 0 {
 				fmt.Println("No results found.")
 				return nil
 			}
-			fmt.Printf("Found %d results for %q (page %d, %d per page):\n\n", len(results), query, page, limit)
+			fmt.Printf("Found %d results for %q — showing %d (page %d, %d per page):\n\n", len(all), query, len(results), page, limit)
 			for _, r := range results {
 				fmt.Printf("  %-8s %-30s [%s] %s\n", shortFP(r.Fingerprint), truncateStr(r.Title, 30), r.Project, r.SourceID)
 				if r.Summary != "" {
@@ -105,13 +111,20 @@ func NewRecentCmd(kdb *db.KnoxDB) *cobra.Command {
 		Use:   "recent",
 		Short: "Show recent knowledge entries",
 		RunE: func(c *cobra.Command, args []string) error {
-			all, err := kdb.RecentEntries(limit * 10)
+			if page < 1 || limit < 1 {
+				return fmt.Errorf("page and limit must be >= 1 (got page=%d limit=%d)", page, limit)
+			}
+			all, err := kdb.RecentEntries(page * limit)
 			if err != nil {
 				return err
 			}
 			entries := paginate(all, page, limit)
 			if len(entries) == 0 {
-				fmt.Println("No entries yet. Run `knox watch` to start ingesting.")
+				if page > 1 {
+					fmt.Printf("No more entries on page %d (page %d of %d).\n", page, page, (len(all)+limit-1)/limit)
+				} else {
+					fmt.Println("No entries yet. Run `knox watch` to start ingesting.")
+				}
 				return nil
 			}
 			fmt.Printf("Recent %d entries (page %d, %d per page):\n\n", len(entries), page, limit)

internal/cmd/topics.go

@@ -5,9 +5,9 @@ import (
 	"regexp"
 	"strings"
 
-	"github.com/spf13/cobra"
 	"github.com/david/knox/internal/db"
 	"github.com/david/knox/internal/index"
+	"github.com/spf13/cobra"
 )
 
 func NewTopicsCmd(kdb *db.KnoxDB) *cobra.Command {

internal/cmd/watch.go

@@ -4,9 +4,9 @@ import (
 	"io"
 	"log"
 
-	"github.com/spf13/cobra"
 	"github.com/david/knox/internal/db"
 	"github.com/david/knox/internal/watch"
+	"github.com/spf13/cobra"
 )
 
 func NewWatchCmd(kdb *db.KnoxDB) *cobra.Command {

internal/db/db.go

@@ -22,6 +22,12 @@ type KnoxDB struct {
 	db     *sql.DB
 	nodeID string
 	clock  *hlc.Clock
+
+	// writeMu serializes the check-then-insert dedup in RecordObservation within
+	// this process. Cross-process serialization comes from _txlock=immediate (the
+	// write lock is taken at BEGIN, before the dedup read) plus a single
+	// connection per pool.
+	writeMu sync.Mutex
 }
 
 type Observation struct {
@@ -93,10 +99,13 @@ func Open(path string) (*KnoxDB, error) {
 		return nil, fmt.Errorf("create db dir: %w", err)
 	}
 
-	db, err := sql.Open("sqlite", path+"?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)")
+	db, err := sql.Open("sqlite", path+"?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)&_txlock=immediate")
 	if err != nil {
 		return nil, fmt.Errorf("open db: %w", err)
 	}
+	// One connection per process: WAL has a single writer; serializing on one
+	// connection avoids pool contention surfacing as busy_timeout errors.
+	db.SetMaxOpenConns(1)
 
 	if _, err := db.Exec(Schema); err != nil {
 		return nil, fmt.Errorf("init schema: %w", err)
@@ -135,6 +144,13 @@ func Open(path string) (*KnoxDB, error) {
 	if _, err := db.Exec("UPDATE observations SET hcl=id WHERE hcl IS NULL"); err != nil {
 		return nil, fmt.Errorf("backfill hcl: %w", err)
 	}
+	// Resume this node's HLC from its persisted max: a restart with a regressed
+	// wall clock must not reissue already-persisted values (see hlc.SeekTo).
+	var maxHCL int64
+	if err := db.QueryRow(`SELECT COALESCE(MAX(hcl), 0) FROM observations WHERE node_id=?`, nodeID).Scan(&maxHCL); err != nil {
+		return nil, fmt.Errorf("seed hlc: %w", err)
+	}
+	kdb.clock.SeekTo(maxHCL)
 
 	// 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.
@@ -142,6 +158,12 @@ func Open(path string) (*KnoxDB, error) {
 		return nil, fmt.Errorf("create locator index: %w", err)
 	}
 
+	// Thread idempotency index must come after the cluster_key migration (a
+	// fresh DB has the column from Schema; an existing DB gets it above).
+	if _, err := db.Exec(`CREATE UNIQUE INDEX IF NOT EXISTS idx_threads_cluster ON threads(cluster_key) WHERE cluster_key IS NOT NULL AND cluster_key != ''`); err != nil {
+		return nil, fmt.Errorf("create cluster_key index: %w", err)
+	}
+
 	return kdb, nil
 }
 
@@ -198,6 +220,9 @@ func (k *KnoxDB) ObservationEntryEstimate() int {
 // Idempotent: if the latest observation for this fingerprint has an identical
 // content signature, nothing is recorded — re-ingesting unchanged content is a no-op.
 func (k *KnoxDB) RecordObservation(o ObservationRecord) (obsID int64, isNew bool, err error) {
+	k.writeMu.Lock()
+	defer k.writeMu.Unlock()
+
 	tx, err := k.db.Begin()
 	if err != nil {
 		return 0, false, fmt.Errorf("begin tx: %w", err)

internal/db/db_test.go

@@ -0,0 +1,156 @@
+package db
+
+import (
+	"path/filepath"
+	"sync"
+	"testing"
+)
+
+func tmpDB(t *testing.T) *KnoxDB {
+	t.Helper()
+	kdb, err := Open(filepath.Join(t.TempDir(), "index.db"))
+	if err != nil {
+		t.Fatalf("open db: %v", err)
+	}
+	t.Cleanup(func() { kdb.Close() })
+	return kdb
+}
+
+func record(t *testing.T, k *KnoxDB, fp, title string) {
+	t.Helper()
+	_, _, err := k.RecordObservation(ObservationRecord{
+		Fingerprint:     fp,
+		SourceID:        "test",
+		SourcePath:      fp,
+		Project:         "itest",
+		ContentType:     "test",
+		Title:           title,
+		Summary:         "summary",
+		CreatedAt:       "2026-08-29T00:00:00Z",
+		LineEnd:         0,
+		Confidence:      0.9,
+		IngesterVersion: "test/v1",
+	})
+	if err != nil {
+		t.Errorf("record %s: %v", fp, err)
+	}
+}
+
+func obsCount(t *testing.T, k *KnoxDB) int {
+	t.Helper()
+	stats, err := k.Stats()
+	if err != nil {
+		t.Fatalf("stats: %v", err)
+	}
+	n, _ := stats["total_observations"].(int)
+	return n
+}
+
+// TestRecordObservationConcurrentDedup: N goroutines ingesting identical
+// content must produce exactly one observation row. This exercises the
+// check-then-insert dedup under the writeMu + BEGIN IMMEDIATE serialization.
+func TestRecordObservationConcurrentDedup(t *testing.T) {
+	k := tmpDB(t)
+	var wg sync.WaitGroup
+	for i := 0; i < 20; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			record(t, k, "fp:concurrent", "same title")
+		}()
+	}
+	wg.Wait()
+	if n := obsCount(t, k); n != 1 {
+		t.Fatalf("expected exactly 1 observation after concurrent identical ingests, got %d", n)
+	}
+}
+
+// TestRecordObservationDistinctFingerprints: different content must never be
+// deduped away (the constraint is per-fingerprint signature, not global).
+func TestRecordObservationDistinctFingerprints(t *testing.T) {
+	k := tmpDB(t)
+	record(t, k, "fp:a", "alpha")
+	record(t, k, "fp:b", "beta")
+	record(t, k, "fp:a", "alpha changed")
+	if n := obsCount(t, k); n != 3 {
+		t.Fatalf("expected 3 observations, got %d", n)
+	}
+}
+
+// TestPushObservationsValidation: forged/malformed rows are rejected without
+// error — self node_id (poisoning vector), empty node_id, negative HCL.
+func TestPushObservationsValidation(t *testing.T) {
+	k := tmpDB(t)
+	foreign := GossipObservation{
+		NodeID: "0123456789abcdef0123456789abcdef", HCL: 42,
+		Fingerprint: "fp:foreign", SourceID: "test", Title: "t", Summary: "s",
+		CollectedAt: "2026-08-29T00:00:00Z",
+	}
+	rows := []GossipObservation{
+		foreign,
+		{NodeID: k.NodeID(), HCL: 100, Fingerprint: "fp:self"}, // spoof poisoning attempt
+		{NodeID: "", HCL: 1, Fingerprint: "fp:empty"},
+		{NodeID: "other", HCL: -5, Fingerprint: "fp:neg"},
+	}
+	n, err := k.PushObservations(rows)
+	if err != nil {
+		t.Fatalf("push: %v", err)
+	}
+	if n != 1 {
+		t.Fatalf("expected exactly the one valid row inserted, got %d", n)
+	}
+	if got := obsCount(t, k); got != 1 {
+		t.Fatalf("expected 1 observation in the log, got %d", got)
+	}
+}
+
+// TestOpenReopenHCLMonotonicAcrossRestart: reopening a DB must resume the HLC
+// from its persisted max (clock seeding), keep the node identity, and order the
+// new observation above every previous one.
+func TestOpenReopenHCLMonotonicAcrossRestart(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "index.db")
+
+	k1, err := Open(path)
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	record(t, k1, "fp:r1", "one")
+	record(t, k1, "fp:r2", "two")
+	record(t, k1, "fp:r3", "three")
+	nodeID1 := k1.NodeID()
+
+	maxBefore := int64(0)
+	rows, err := k1.ObservationsAfter(nodeID1, 0, 100)
+	if err != nil {
+		t.Fatalf("obs after: %v", err)
+	}
+	for _, r := range rows {
+		if r.HCL > maxBefore {
+			maxBefore = r.HCL
+		}
+	}
+	if err := k1.Close(); err != nil {
+		t.Fatalf("close: %v", err)
+	}
+
+	k2, err := Open(path)
+	if err != nil {
+		t.Fatalf("reopen: %v", err)
+	}
+	defer k2.Close()
+	if k2.NodeID() != nodeID1 {
+		t.Errorf("node id changed across reopen: %q -> %q", nodeID1, k2.NodeID())
+	}
+	record(t, k2, "fp:r4", "four")
+
+	after, err := k2.ObservationsAfter(nodeID1, maxBefore, 100)
+	if err != nil {
+		t.Fatalf("obs after (reopened): %v", err)
+	}
+	if len(after) != 1 {
+		t.Fatalf("expected exactly the new observation above the pre-restart max, got %d rows", len(after))
+	}
+	if after[0].Fingerprint != "fp:r4" {
+		t.Errorf("unexpected row above max: %s", after[0].Fingerprint)
+	}
+}

internal/db/gossip.go

@@ -3,6 +3,7 @@ package db
 import (
 	"database/sql"
 	"fmt"
+	"log"
 )
 
 // GossipObservation is the serializable wire form of an observation exchanged
@@ -40,6 +41,19 @@ func (k *KnoxDB) PushObservations(rows []GossipObservation) (int, error) {
 
 	inserted := 0
 	for _, o := range rows {
+		// Reject malformed or forged rows. Nodes only ever push their own
+		// observations, so a row claiming this node's id cannot be legitimate:
+		// accepting it would let a peer poison our knowledge vector (a forged
+		// max-HCL makes peers believe they have our whole history and stop
+		// pulling). Empty node ids and negative HCLs are likewise never produced
+		// by a real node.
+		if o.NodeID == "" || o.HCL < 0 {
+			continue
+		}
+		if o.NodeID == k.nodeID {
+			log.Printf("[gossip] dropped pushed row claiming local node_id (spoof?)")
+			continue
+		}
 		res, err := tx.Exec(
 			`INSERT OR IGNORE INTO observations
 			 (fingerprint, source_id, source_path, project, content_type, title, summary,
@@ -141,6 +155,18 @@ func (k *KnoxDB) UpsertPeer(peerID, addr, name string, maxHCL int64) error {
 	return err
 }
 
+// MergePeer records a peer discovered indirectly (via another peer's ping).
+// Unlike UpsertPeer it never clobbers the cursor — a freshly learned address
+// has no known knowledge yet; the anti-entropy pull will set it on contact.
+func (k *KnoxDB) MergePeer(peerID, addr, name string) error {
+	_, err := k.db.Exec(
+		`INSERT INTO peers (peer_id, addr, name) VALUES (?, ?, ?)
+		 ON CONFLICT(peer_id) DO UPDATE SET addr=?, name=?`,
+		peerID, addr, name, addr, name,
+	)
+	return err
+}
+
 // ListPeers returns known peers ordered by first-seen.
 func (k *KnoxDB) ListPeers() ([]Peer, error) {
 	rows, err := k.db.Query(`SELECT peer_id, COALESCE(addr,''), COALESCE(name,''), COALESCE(last_handshake,''), COALESCE(cursor,0), COALESCE(created_at,'') FROM peers ORDER BY created_at`)
@@ -181,6 +207,61 @@ type Peer struct {
 	CreatedAt     string
 }
 
+// PeerInfo is the shareable (non-secret) subset of a peer that /v1/ping
+// advertises so other nodes can discover the swarm.
+type PeerInfo struct {
+	PeerID string `json:"peer_id"`
+	Addr   string `json:"addr"`
+	Name   string `json:"name"`
+}
+
+// ShareablePeers returns the peers this node knows about, for dissemination in
+// ping responses. Self and peers without an address are excluded.
+func (k *KnoxDB) ShareablePeers() ([]PeerInfo, error) {
+	rows, err := k.db.Query(
+		`SELECT peer_id, COALESCE(addr,''), COALESCE(name,'') FROM peers WHERE addr<>'' AND peer_id<>? ORDER BY peer_id`,
+		k.NodeID(),
+	)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var out []PeerInfo
+	for rows.Next() {
+		var p PeerInfo
+		if err := rows.Scan(&p.PeerID, &p.Addr, &p.Name); err != nil {
+			return nil, err
+		}
+		out = append(out, p)
+	}
+	return out, rows.Err()
+}
+
+// MaxHCLForNode returns the maximum HCL this node holds for a given origin
+// node, or 0 if none.
+func (k *KnoxDB) MaxHCLForNode(nodeID string) int64 {
+	var m int64
+	if err := k.db.QueryRow(`SELECT MAX(hcl) FROM observations WHERE node_id=?`, nodeID).Scan(&m); err != nil {
+		return 0
+	}
+	return m
+}
+
+// SwarmPeerAddrs returns the addresses of all known peers (shareable set). It
+// is what the sweep iterates after a bootstrap join.
+func (k *KnoxDB) SwarmPeerAddrs() ([]string, error) {
+	peers, err := k.ShareablePeers()
+	if err != nil {
+		return nil, err
+	}
+	addrs := make([]string, 0, len(peers))
+	for _, p := range peers {
+		addrs = append(addrs, p.Addr)
+	}
+	return addrs, nil
+}
+
 // DistinctFingerprints returns the set of all observed fingerprints — the
 // ground-truth index of what this node knows. Used by gossip diff.
 func (k *KnoxDB) DistinctFingerprints() (map[string]bool, error) {
@@ -219,4 +300,4 @@ func (k *KnoxDB) ThreadStatusByCluster() (map[string]string, error) {
 		out[key] = status
 	}
 	return out, rows.Err()
-}
+}

internal/db/metrics.go

@@ -0,0 +1,108 @@
+package db
+
+// MetricsSnapshot holds the scrape-time gauges derived from the database.
+type MetricsSnapshot struct {
+	Observations        int
+	ObservationsLast24h int
+	BySource            map[string]int
+	Entries             int
+	Projects            int
+	Sessions            int
+	PendingReflections  int
+	ThreadsByStatus     map[string]int
+	Peers               int
+	ByOriginNode        map[string]int   // node_id → observation count
+	KnowledgeVector     map[string]int64 // node_id → max hcl
+	EarliestObservation string
+}
+
+// MetricsSnapshot computes database-derived gauges for Prometheus scraping.
+// All queries are cheap aggregations; nothing is stored or mutated.
+func (k *KnoxDB) MetricsSnapshot() (*MetricsSnapshot, error) {
+	s := &MetricsSnapshot{
+		BySource:        make(map[string]int),
+		ThreadsByStatus: make(map[string]int),
+		ByOriginNode:    make(map[string]int),
+	}
+
+	if err := k.db.QueryRow("SELECT COUNT(*) FROM observations").Scan(&s.Observations); err != nil {
+		return nil, err
+	}
+	if err := k.db.QueryRow("SELECT COUNT(*) FROM observations WHERE collected_at > datetime('now', '-1 day')").Scan(&s.ObservationsLast24h); err != nil {
+		return nil, err
+	}
+	if err := k.db.QueryRow("SELECT COUNT(*) FROM entries").Scan(&s.Entries); err != nil {
+		return nil, err
+	}
+	if err := k.db.QueryRow("SELECT COUNT(DISTINCT project) FROM entries").Scan(&s.Projects); err != nil {
+		return nil, err
+	}
+	if err := k.db.QueryRow("SELECT COUNT(*) FROM sessions").Scan(&s.Sessions); err != nil {
+		return nil, err
+	}
+	if err := k.db.QueryRow("SELECT COUNT(*) FROM sessions WHERE indexed=0").Scan(&s.PendingReflections); err != nil {
+		return nil, err
+	}
+	if err := k.db.QueryRow("SELECT COUNT(*) FROM peers").Scan(&s.Peers); err != nil {
+		return nil, err
+	}
+	if err := k.db.QueryRow("SELECT COALESCE(MIN(collected_at),'') FROM observations").Scan(&s.EarliestObservation); err != nil {
+		return nil, err
+	}
+
+	rows, err := k.db.Query("SELECT COALESCE(source_id,'unknown'), COUNT(*) FROM observations GROUP BY source_id")
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var src string
+		var n int
+		if err := rows.Scan(&src, &n); err != nil {
+			return nil, err
+		}
+		s.BySource[src] = n
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+
+	rows2, err := k.db.Query("SELECT COALESCE(status,'active'), COUNT(*) FROM threads GROUP BY status")
+	if err != nil {
+		return nil, err
+	}
+	defer rows2.Close()
+	for rows2.Next() {
+		var st string
+		var n int
+		if err := rows2.Scan(&st, &n); err != nil {
+			return nil, err
+		}
+		s.ThreadsByStatus[st] = n
+	}
+	if err := rows2.Err(); err != nil {
+		return nil, err
+	}
+
+	rows3, err := k.db.Query("SELECT node_id, COUNT(*) FROM observations WHERE node_id<>'' GROUP BY node_id")
+	if err != nil {
+		return nil, err
+	}
+	defer rows3.Close()
+	for rows3.Next() {
+		var nid string
+		var n int
+		if err := rows3.Scan(&nid, &n); err != nil {
+			return nil, err
+		}
+		s.ByOriginNode[nid] = n
+	}
+	if err := rows3.Err(); err != nil {
+		return nil, err
+	}
+
+	if s.KnowledgeVector, err = k.KnowledgeVector(); err != nil {
+		return nil, err
+	}
+	return s, nil
+}

internal/db/schema.go

@@ -105,8 +105,6 @@ CREATE TABLE IF NOT EXISTS threads (
     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/hlc/hlc.go

@@ -27,6 +27,23 @@ type Clock struct {
 
 func New() *Clock { return &Clock{} }
 
+// SeekTo adopts the given packed HLC value when it is ahead of the clock's current
+// position, so the next Now is still strictly increasing. Used to resume a node's
+// clock from its persisted MAX(hcl) at startup — without it, a restart with a
+// regressed wall clock would reissue already-used values and break the
+// monotonicity the (node_id, hcl) locator uniqueness and gossip cursors rely on.
+func (c *Clock) SeekTo(v int64) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	wall := v >> wallShift
+	seq := v & seqMask
+	if wall > c.wallMS || (wall == c.wallMS && seq > c.seq) {
+		c.wallMS = wall
+		c.seq = seq
+	}
+}
+
 // Now returns the next monotonic HLC value and the wall-clock time embedded in
 // it. The returned time is the HLC's wall component — never ahead of the local
 // clock beyond the current call and never rewinding across calls.
@@ -57,4 +74,4 @@ func (c *Clock) Now() (int64, time.Time) {
 // WallTime extracts the wall-clock component embedded in a packed HLC value.
 func WallTime(v int64) time.Time {
 	return time.UnixMilli(v >> wallShift).UTC()
-}
+}

internal/hlc/hlc_test.go

@@ -0,0 +1,68 @@
+package hlc
+
+import "testing"
+
+// TestSeekToFutureValueKeepsMonotonic: after resuming from a persisted value
+// ahead of the wall clock (clock regression), every subsequent Now must still
+// be strictly greater than the resumed position.
+func TestSeekToFutureValueKeepsMonotonic(t *testing.T) {
+	c := New()
+	resumed := int64(1) << 62 // packed value far ahead of any real wall clock
+	c.SeekTo(resumed)
+
+	prev, _ := c.Now()
+	if prev <= resumed {
+		t.Fatalf("first Now after SeekTo = %d, want > resumed %d", prev, resumed)
+	}
+	for i := 0; i < 100; i++ {
+		next, _ := c.Now()
+		if next <= prev {
+			t.Fatalf("HLC regressed: %d then %d", prev, next)
+		}
+		prev = next
+	}
+}
+
+// TestSeekToLowerValueIgnored: resuming from a value behind the current clock
+// (or a fresh 0-padded DB) must not rewind it.
+func TestSeekToLowerValueIgnored(t *testing.T) {
+	c := New()
+	first, _ := c.Now()
+	c.SeekTo(0)
+	second, _ := c.Now()
+	if second <= first {
+		t.Fatalf("SeekTo(0) rewound the clock: %d then %d", first, second)
+	}
+
+	// Seek to exactly the last emitted value: the next value must exceed it.
+	c.SeekTo(second)
+	third, _ := c.Now()
+	if third <= second {
+		t.Fatalf("SeekTo(last) did not preserve monotonicity: %d then %d", second, third)
+	}
+}
+
+// TestSeekToAcrossRestart mirrors Open's reopen path: a fresh clock resumed
+// from the persisted max keeps issuing strictly increasing values.
+func TestSeekToAcrossRestart(t *testing.T) {
+	c1 := New()
+	var last int64
+	for i := 0; i < 50; i++ {
+		last, _ = c1.Now()
+	}
+
+	c2 := New() // fresh process clock
+	c2.SeekTo(last)
+
+	prev, _ := c2.Now()
+	if prev <= last {
+		t.Fatalf("reopened clock reissued a value: %d <= %d", prev, last)
+	}
+	for i := 0; i < 50; i++ {
+		next, _ := c2.Now()
+		if next <= prev {
+			t.Fatalf("reopened clock regressed: %d then %d", prev, next)
+		}
+		prev = next
+	}
+}

internal/metrics/metrics.go

@@ -0,0 +1,164 @@
+// Package metrics exposes Prometheus-format metrics for a knox node.
+//
+// Gauges are recomputed from the database on each scrape (cheap aggregates);
+// gossip counters are in-memory and incremented as the daemon exchanges data
+// with peers. The registry also gains the standard Go runtime and process
+// collectors from prometheus/client_golang.
+package metrics
+
+import (
+	"net/http"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+
+	"github.com/david/knox/internal/db"
+)
+
+// Metrics holds the gossip event counters (incremented by watch) and the
+// scrape-time gauges derived from the database (refreshed on each scrape).
+type Metrics struct {
+	// Gossip counters (live).
+	pullsTotal     prometheus.Counter
+	pushesTotal    prometheus.Counter
+	obsPulledTotal prometheus.Counter
+	obsPushedTotal prometheus.Counter
+	errorsTotal    prometheus.Counter
+
+	// Snapshot gauges (updated per scrape).
+	observationsGauge   *prometheus.GaugeVec
+	entriesGauge        prometheus.Gauge
+	projectsGauge       prometheus.Gauge
+	sessionsGauge       prometheus.Gauge
+	pendingReflections  prometheus.Gauge
+	peersGauge          prometheus.Gauge
+	threadsByStatus     *prometheus.GaugeVec
+	byOriginNode        *prometheus.GaugeVec
+	knowledgeVector     *prometheus.GaugeVec
+	observationsLast24h prometheus.Gauge
+
+	registry *prometheus.Registry
+	kdb      *db.KnoxDB
+}
+
+// New builds the metrics registry bound to a knowledge index.
+func New(kdb *db.KnoxDB, nodeName string) *Metrics {
+	reg := prometheus.NewRegistry()
+
+	m := &Metrics{
+		registry: reg,
+		kdb:      kdb,
+	}
+
+	// Node identity aids scraping: which node produced this output.
+	m.nodeInfo(kdb.NodeID(), nodeName)
+
+	m.pullsTotal = newCounter(reg, "knox_gossip_pulls_total", "Peer pull round-trips completed.")
+	m.pushesTotal = newCounter(reg, "knox_gossip_pushes_total", "Peer push round-trips completed.")
+	m.obsPulledTotal = newCounter(reg, "knox_gossip_observations_pulled_total", "Observations received from peers.")
+	m.obsPushedTotal = newCounter(reg, "knox_gossip_observations_pushed_total", "Observations sent to peers.")
+	m.errorsTotal = newCounter(reg, "knox_gossip_errors_total", "Gossip errors (ping/pull/push failures).")
+
+	m.observationsGauge = newGaugeVec(reg, "knox_observations_total", "Observation log size.", "source_id")
+	m.observationsLast24h = newGauge(reg, "knox_observations_last_24h", "Observations collected in the last 24h.")
+	m.entriesGauge = newGauge(reg, "knox_entries_total", "Materialized entry cache size.")
+	m.projectsGauge = newGauge(reg, "knox_projects_total", "Distinct projects in the entry cache.")
+	m.sessionsGauge = newGauge(reg, "knox_sessions_total", "Sessions tracked.")
+	m.pendingReflections = newGauge(reg, "knox_pending_reflections", "Sessions awaiting reflection.")
+	m.peersGauge = newGauge(reg, "knox_peers_total", "Known peer nodes.")
+	m.threadsByStatus = newGaugeVec(reg, "knox_threads_total", "Threads by status.", "status")
+	m.byOriginNode = newGaugeVec(reg, "knox_observations_by_node", "Observations per originating node.", "node_id")
+	m.knowledgeVector = newGaugeVec(reg, "knox_knowledge_max_hcl", "Highest HCL seen per originating node.", "node_id")
+
+	// Go runtime + process collectors come from the official library.
+	reg.MustRegister(prometheus.NewGoCollector())
+	reg.MustRegister(prometheus.NewProcessCollector(prometheus.ProcessCollectorOpts{}))
+
+	return m
+}
+
+func newCounter(reg *prometheus.Registry, name, help string) prometheus.Counter {
+	c := prometheus.NewCounter(prometheus.CounterOpts{Name: name, Help: help})
+	reg.MustRegister(c)
+	return c
+}
+
+func newGauge(reg *prometheus.Registry, name, help string) prometheus.Gauge {
+	g := prometheus.NewGauge(prometheus.GaugeOpts{Name: name, Help: help})
+	reg.MustRegister(g)
+	return g
+}
+
+func newGaugeVec(reg *prometheus.Registry, name, help string, labels ...string) *prometheus.GaugeVec {
+	g := prometheus.NewGaugeVec(prometheus.GaugeOpts{Name: name, Help: help}, labels)
+	reg.MustRegister(g)
+	return g
+}
+
+func (m *Metrics) nodeInfo(nodeID, name string) {
+	info := prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "knox_node_info",
+		Help: "Node identity (always 1).",
+		ConstLabels: prometheus.Labels{
+			"node_id": nodeID,
+			"name":    name,
+		},
+	})
+	m.registry.MustRegister(info)
+	info.Set(1)
+}
+
+// Capture refresh the DB-derived gauges from a fresh snapshot.
+func (m *Metrics) Capture(s *db.MetricsSnapshot) {
+	m.observationsGauge.Reset()
+	for src, n := range s.BySource {
+		m.observationsGauge.WithLabelValues(src).Set(float64(n))
+	}
+	m.observationsLast24h.Set(float64(s.ObservationsLast24h))
+	m.entriesGauge.Set(float64(s.Entries))
+	m.projectsGauge.Set(float64(s.Projects))
+	m.sessionsGauge.Set(float64(s.Sessions))
+	m.pendingReflections.Set(float64(s.PendingReflections))
+	m.peersGauge.Set(float64(s.Peers))
+
+	m.threadsByStatus.Reset()
+	for st, n := range s.ThreadsByStatus {
+		m.threadsByStatus.WithLabelValues(st).Set(float64(n))
+	}
+
+	m.byOriginNode.Reset()
+	for nid, n := range s.ByOriginNode {
+		m.byOriginNode.WithLabelValues(nid).Set(float64(n))
+	}
+	m.knowledgeVector.Reset()
+	for nid, hcl := range s.KnowledgeVector {
+		m.knowledgeVector.WithLabelValues(nid).Set(float64(hcl))
+	}
+}
+
+// IncrementPull records a completed pull and its accepted observation count.
+func (m *Metrics) IncrementPull(newObs int) {
+	m.pullsTotal.Inc()
+	m.obsPulledTotal.Add(float64(newObs))
+}
+
+// IncrementPush records a completed push and its accepted observation count.
+func (m *Metrics) IncrementPush(newObs int) {
+	m.pushesTotal.Inc()
+	m.obsPushedTotal.Add(float64(newObs))
+}
+
+// IncrementErrors counts a failed gossip attempt.
+func (m *Metrics) IncrementErrors() { m.errorsTotal.Inc() }
+
+// Handler returns the /metrics scrape handler. On each scrape it refreshes
+// DB-derived gauges before rendering (cost is a few cheap aggregates).
+func (m *Metrics) Handler() http.Handler {
+	h := promhttp.HandlerFor(m.registry, promhttp.HandlerOpts{})
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if s, err := m.kdb.MetricsSnapshot(); err == nil {
+			m.Capture(s)
+		}
+		h.ServeHTTP(w, r)
+	})
+}

internal/metrics/metrics_test.go

@@ -0,0 +1,114 @@
+package metrics
+
+import (
+	"io"
+	"net/http/httptest"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/david/knox/internal/db"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+)
+
+func tmpKdb(t *testing.T) *db.KnoxDB {
+	t.Helper()
+	k, err := db.Open(filepath.Join(t.TempDir(), "index.db"))
+	if err != nil {
+		t.Fatalf("open db: %v", err)
+	}
+	t.Cleanup(func() { k.Close() })
+	return k
+}
+
+func seed(t *testing.T, k *db.KnoxDB, src string, n int) {
+	t.Helper()
+	for i := 0; i < n; i++ {
+		_, _, err := k.RecordObservation(db.ObservationRecord{
+			Fingerprint:     "fp-" + src + "-" + string(rune('a'+i)),
+			SourceID:        src,
+			SourcePath:      src,
+			Project:         "test",
+			ContentType:     "test",
+			Title:           src,
+			Summary:         "s",
+			CreatedAt:       "2026-08-29T00:00:00Z",
+			Confidence:      0.9,
+			IngesterVersion: "itest/v1",
+		})
+		if err != nil {
+			t.Fatalf("seed: %v", err)
+		}
+	}
+}
+
+func TestMetricsSnapshot(t *testing.T) {
+	k := tmpKdb(t)
+	seed(t, k, "git", 2)
+	seed(t, k, "browser-history", 3)
+
+	s, err := k.MetricsSnapshot()
+	if err != nil {
+		t.Fatalf("snapshot: %v", err)
+	}
+	if s.Observations != 5 {
+		t.Errorf("observations = %d, want 5", s.Observations)
+	}
+	if s.BySource["git"] != 2 || s.BySource["browser-history"] != 3 {
+		t.Errorf("by source = %v", s.BySource)
+	}
+	if s.KnowledgeVector[k.NodeID()] == 0 {
+		t.Errorf("knowledge vector missing own node")
+	}
+	if s.ByOriginNode[k.NodeID()] != 5 {
+		t.Errorf("by origin node = %v", s.ByOriginNode)
+	}
+}
+
+func TestMetricsScrape(t *testing.T) {
+	k := tmpKdb(t)
+	seed(t, k, "git", 2)
+
+	m := New(k, "testnode")
+	h := m.Handler()
+
+	rec := httptest.NewRecorder()
+	h.ServeHTTP(rec, httptest.NewRequest("GET", "/metrics", nil))
+
+	body, _ := io.ReadAll(rec.Body)
+	out := string(body)
+	for _, want := range []string{
+		`knox_node_info{name="testnode"`,
+		`knox_observations_total{source_id="git"} 2`,
+		`knox_gossip_pulls_total 0`,
+		"go_goroutines",
+		"process_cpu_seconds_total",
+	} {
+		if !strings.Contains(out, want) {
+			t.Errorf("scrape output missing %q", want)
+		}
+	}
+}
+
+func TestMetricsCounters(t *testing.T) {
+	k := tmpKdb(t)
+	m := New(k, "t")
+
+	// Manually drive counters through the Metrics API.
+	m.IncrementPull(3)
+	m.IncrementPush(7)
+	m.IncrementErrors()
+
+	if got := testutil.ToFloat64(m.pullsTotal); got != 1 {
+		t.Errorf("pullsTotal = %v, want 1", got)
+	}
+	if got := testutil.ToFloat64(m.obsPulledTotal); got != 3 {
+		t.Errorf("obsPulled = %v, want 3", got)
+	}
+	if got := testutil.ToFloat64(m.obsPushedTotal); got != 7 {
+		t.Errorf("obsPushed = %v, want 7", got)
+	}
+	if got := testutil.ToFloat64(m.errorsTotal); got != 1 {
+		t.Errorf("errorsTotal = %v, want 1", got)
+	}
+}

internal/watch/gossip.go

@@ -19,22 +19,25 @@ import (
 	"time"
 
 	"github.com/david/knox/internal/db"
+	"github.com/david/knox/internal/metrics"
 )
 
 const defaultPort = "8931"
 
 type Node struct {
-	Kdb  *db.KnoxDB
-	Name string
-	Addr string // advertised base URL, e.g. http://192.168.1.20:8931
+	Kdb     *db.KnoxDB
+	Name    string
+	Addr    string // advertised base URL, e.g. http://192.168.1.20:8931
+	Metrics *metrics.Metrics
 }
 
 // pingResponse is the anti-entropy summary returned by /v1/ping.
 type pingResponse struct {
-	NodeID   string            `json:"node_id"`
-	Name     string            `json:"name"`
-	Vector   map[string]int64  `json:"vector"` // node_id → max hcl
-	MaxHCL   *int64            `json:"max_hcl,omitempty"`
+	NodeID string           `json:"node_id"`
+	Name   string           `json:"name"`
+	Vector map[string]int64 `json:"vector"` // node_id → max hcl
+	Peers  []db.PeerInfo    `json:"peers"`  // swarm membership this node knows
+	MaxHCL *int64           `json:"max_hcl,omitempty"`
 }
 
 func (n *Node) Handler() http.Handler {
@@ -52,10 +55,16 @@ func (n *Node) handlePing(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
+	peers, err := n.Kdb.ShareablePeers()
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
 	writeJSON(w, pingResponse{
 		NodeID: n.Kdb.NodeID(),
 		Name:   n.Name,
 		Vector: vector,
+		Peers:  peers,
 	})
 }
 
@@ -89,10 +98,16 @@ func nextCursor(rows []db.GossipObservation) int64 {
 	return rows[len(rows)-1].HCL
 }
 
+// maxBatchRows bounds the number of observations a peer may push in one POST.
+// Pull already pages at 500 rows, so any larger batch is at best redundant and
+// at worst a flood; capping keeps memory and insert work bounded.
+const maxBatchRows = 1000
+
 func (n *Node) handleBatch(w http.ResponseWriter, r *http.Request) {
+	r.Body = http.MaxBytesReader(w, r.Body, 4<<20) // 4 MiB
 	body, err := io.ReadAll(r.Body)
 	if err != nil {
-		http.Error(w, err.Error(), http.StatusBadRequest)
+		http.Error(w, "batch too large or unreadable", http.StatusBadRequest)
 		return
 	}
 	var rows []db.GossipObservation
@@ -100,6 +115,10 @@ func (n *Node) handleBatch(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, "bad json: "+err.Error(), http.StatusBadRequest)
 		return
 	}
+	if len(rows) > maxBatchRows {
+		http.Error(w, fmt.Sprintf("batch too large: %d rows (max %d)", len(rows), maxBatchRows), http.StatusBadRequest)
+		return
+	}
 	inserted, err := n.Kdb.PushObservations(rows)
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
@@ -114,11 +133,11 @@ func (n *Node) handleBatch(w http.ResponseWriter, r *http.Request) {
 // DiffSummary is the divergence snapshot served at /v1/diff: everything this
 // node observes, plus the status of auto-threaded threads (for tombstones).
 type DiffSummary struct {
-	NodeID        string            `json:"node_id"`
-	Name          string            `json:"name"`
-	Fingerprints  []string          `json:"fingerprints"`
-	ThreadStatus  map[string]string `json:"thread_status"`
-	ObsCount      int               `json:"obs_count"`
+	NodeID       string            `json:"node_id"`
+	Name         string            `json:"name"`
+	Fingerprints []string          `json:"fingerprints"`
+	ThreadStatus map[string]string `json:"thread_status"`
+	ObsCount     int               `json:"obs_count"`
 }
 
 func (n *Node) handleDiff(w http.ResponseWriter, r *http.Request) {
@@ -154,9 +173,9 @@ func writeJSON(w http.ResponseWriter, v any) {
 
 // Client is the pull/push half used by the exchange loop.
 type Client struct {
-	NodeID   string
-	Addr     string
-	Timeout  time.Duration
+	NodeID  string
+	Addr    string
+	Timeout time.Duration
 }
 
 func (c *Client) Ping() (*pingResponse, error) {
@@ -255,27 +274,63 @@ func (c *Client) Diff() (*DiffSummary, error) {
 	return &d, nil
 }
 
-// Run executes one anti-entropy sweep against the given peer addresses.
-func Run(kdb *db.KnoxDB, peers []string) {
+// Run executes one anti-entropy + membership sweep.
+//
+// Peers are a fusion of the statically configured list (KNOX_PEERS) and peers
+// previously discovered and persisted in the peers table (swarm join): syncing
+// with any one node reveals who else is in the swarm, and those nodes are then
+// swept too. There is no relay of observations — only membership is shared; each
+// node pulls/pushes directly with every other node it learns about.
+//
+// m, when non-nil, receives gossip event counters (nil for one-shot CLI runs).
+func Run(kdb *db.KnoxDB, m *metrics.Metrics, static []string) int {
 	myID := kdb.NodeID()
-	for _, addr := range peers {
-		addr = strings.TrimSpace(addr)
-		if addr == "" {
-			continue
-		}
-		if strings.HasPrefix(addr, kdb.NodeID()+":") {
+
+	// Seed the work queue with static config plus persisted discoveries.
+	persisted, _ := kdb.SwarmPeerAddrs()
+	work := make([]string, 0, len(static)+len(persisted))
+	work = append(work, static...)
+	work = append(work, persisted...)
+
+	seen := make(map[string]bool) // addr → handled (also suppresses self)
+	queue := 0
+	pulledTotal := 0
+	for queue < len(work) {
+		addr := strings.TrimSpace(work[queue])
+		queue++
+		if addr == "" || seen[addr] {
 			continue
 		}
+		seen[addr] = true
+
 		c := &Client{Addr: addr, Timeout: 10 * time.Second}
 		p, err := c.Ping()
 		if err != nil {
 			log.Printf("[gossip] ping %s: %v", addr, err)
+			if m != nil {
+				m.IncrementErrors()
+			}
 			continue
 		}
 		if p.NodeID == myID {
 			continue // never talk to ourselves (or an aliased address)
 		}
 
+		// Membership discovery: learn who else is in the swarm and enqueue
+		// their addresses for direct sweeps.
+		for _, pi := range p.Peers {
+			if pi.PeerID == "" || pi.PeerID == myID || pi.Addr == "" {
+				continue
+			}
+			if err := kdb.MergePeer(pi.PeerID, pi.Addr, pi.Name); err != nil {
+				log.Printf("[gossip] merge peer %s: %v", pi.PeerID, err)
+				continue
+			}
+			if !seen[pi.Addr] {
+				work = append(work, pi.Addr)
+			}
+		}
+
 		pulled := 0
 		for remoteNode, remoteHCL := range p.Vector {
 			localHCL := kdbVectorGet(kdb, remoteNode)
@@ -286,13 +341,23 @@ func Run(kdb *db.KnoxDB, peers []string) {
 				n, err := c.Pull(remoteNode, localHCL, kdb)
 				if err != nil {
 					log.Printf("[gossip] pull %s@%s: %v", remoteNode, addr, err)
+					if m != nil {
+						m.IncrementErrors()
+					}
 					continue
 				}
 				pulled += n
 			}
 		}
+		pulledTotal += pulled
+		if m != nil {
+			m.IncrementPull(pulled)
+		}
 
 		pushed, _ := c.Push(kdb, p.Vector)
+		if m != nil {
+			m.IncrementPush(pushed)
+		}
 
 		if err := kdb.UpsertPeer(p.NodeID, addr, p.Name, vectorMax(p.Vector)); err != nil {
 			log.Printf("[gossip] peer upsert: %v", err)
@@ -303,6 +368,7 @@ func Run(kdb *db.KnoxDB, peers []string) {
 			log.Printf("[gossip] synced with %s (%s): in sync (pushed %d)", p.NodeID, addr, pushed)
 		}
 	}
+	return pulledTotal
 }
 
 func kdbVectorGet(kdb *db.KnoxDB, nodeID string) int64 {
@@ -331,6 +397,18 @@ func ListenAddr() (addr string) {
 	return "localhost:" + defaultPort
 }
 
+const defaultMetricsPort = "8932"
+
+// MetricsAddr returns the Prometheus scrape address (KNOX_METRICS_ADDR or
+// default). It is a separate port from gossip so scraping never contends with
+// the peer protocol.
+func MetricsAddr() (addr string) {
+	if addr = os.Getenv("KNOX_METRICS_ADDR"); addr != "" {
+		return addr
+	}
+	return "localhost:" + defaultMetricsPort
+}
+
 // PeerAddrs returns the configured peer list (KNOX_PEERS, comma-separated).
 func PeerAddrs() []string {
 	raw := os.Getenv("KNOX_PEERS")
@@ -345,4 +423,4 @@ func PeerAddrs() []string {
 		}
 	}
 	return out
-}
+}

internal/watch/gossip_test.go

@@ -1,8 +1,12 @@
 package watch
 
 import (
+	"bytes"
+	"encoding/json"
+	"net/http"
 	"net/http/httptest"
 	"path/filepath"
+	"strings"
 	"testing"
 
 	"github.com/david/knox/internal/db"
@@ -60,8 +64,8 @@ func TestGossipConvergence(t *testing.T) {
 	defer sb.Close()
 
 	// A pulls from B, then B pulls from A (bidirectional sweep).
-	Run(a, []string{sb.URL})
-	Run(b, []string{sa.URL})
+	Run(a, nil, []string{sb.URL})
+	Run(b, nil, []string{sa.URL})
 
 	av, err := a.KnowledgeVector()
 	if err != nil {
@@ -84,8 +88,77 @@ func TestGossipConvergence(t *testing.T) {
 	}
 }
 
-// TestGossipDiff ensures /v1/diff reports per-node observation fingerprints and
-// tombstoned thread divergence.
+// TestGossipSwarmDiscovery: C only knows A. A knows B. When C sweeps A, it must
+// learn about B through A's ping, enqueue B, and pull B's observations — with no
+// direct configuration of B (no relay of data, only membership).
+func TestGossipSwarmDiscovery(t *testing.T) {
+	a := tmpKnoxDB(t)
+	b := tmpKnoxDB(t)
+	c := tmpKnoxDB(t)
+
+	seedObs(a, "AAA")
+	seedObs(b, "BBB")
+	seedObs(c, "CCC")
+
+	nodeA := &Node{Kdb: a, Name: "A"}
+	sa := httptest.NewServer(nodeA.Handler())
+	defer sa.Close()
+	nodeB := &Node{Kdb: b, Name: "B"}
+	sb := httptest.NewServer(nodeB.Handler())
+	defer sb.Close()
+	nodeC := &Node{Kdb: c, Name: "C"}
+	sc := httptest.NewServer(nodeC.Handler())
+	defer sc.Close()
+
+	// A discovers B (A pings B) so A can advertise B to the swarm.
+	Run(a, nil, []string{sb.URL})
+
+	// C only knows A. A single sweep should surface B (membership in ping)
+	// and pull B's observations directly.
+	Run(c, nil, []string{sa.URL})
+
+	// C must know B and hold all three origin logs.
+	peers, err := c.ListPeers()
+	if err != nil {
+		t.Fatalf("list peers: %v", err)
+	}
+	foundB := false
+	for _, p := range peers {
+		if p.PeerID == b.NodeID() {
+			foundB = true
+		}
+	}
+	if !foundB {
+		t.Fatalf("C did not discover B via A's membership list; peers=%v", peers)
+	}
+
+	vec, err := c.KnowledgeVector()
+	if err != nil {
+		t.Fatalf("c vector: %v", err)
+	}
+	if len(vec) != 3 {
+		t.Errorf("C should hold 3 origin logs (A, B, C), got %v", vec)
+	}
+
+	// C's copy of B's log must match B's own max hcl.
+	bmax := c.MaxHCLForNode(b.NodeID())
+	bm, err := bNodeMax(b)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if bmax != bm {
+		t.Errorf("C max hcl for B=%d, B reports %d", bmax, bm)
+	}
+}
+
+// bNodeMax is a helper reading the max hcl B holds for its own node.
+func bNodeMax(b *db.KnoxDB) (int64, error) {
+	rows, err := b.KnowledgeVector()
+	if err != nil {
+		return 0, err
+	}
+	return rows[b.NodeID()], nil
+}
 func TestGossipDiff(t *testing.T) {
 	a := tmpKnoxDB(t)
 	b := tmpKnoxDB(t)
@@ -158,17 +231,117 @@ func TestGossipIdempotent(t *testing.T) {
 	sb := httptest.NewServer(nodeB.Handler())
 	defer sb.Close()
 
-	Run(b, []string{sa.URL})
-	before, err := b.EntryCount()
+	Run(b, nil, []string{sa.URL})
+	entryBefore, err := b.EntryCount()
 	if err != nil {
 		t.Fatal(err)
 	}
-	Run(b, []string{sa.URL})
-	after, err := b.EntryCount()
+	stats, err := b.Stats()
 	if err != nil {
 		t.Fatal(err)
 	}
-	if before != after {
-		t.Errorf("second sweep changed entry count: %d -> %d", before, after)
+	obsBefore, _ := stats["total_observations"].(int)
+
+	Run(b, nil, []string{sa.URL})
+
+	entryAfter, err := b.EntryCount()
+	if err != nil {
+		t.Fatal(err)
 	}
-}
+	if entryBefore != entryAfter {
+		t.Errorf("second sweep changed entry count: %d -> %d", entryBefore, entryAfter)
+	}
+	stats, err = b.Stats()
+	if err != nil {
+		t.Fatal(err)
+	}
+	obsAfter, _ := stats["total_observations"].(int)
+	if obsBefore != obsAfter {
+		t.Errorf("second sweep duplicated observations: %d -> %d", obsBefore, obsAfter)
+	}
+}
+
+// TestHandleBatchRejectsOversizedBatch: more than maxBatchRows in one POST
+// must be refused up front, before any insert work.
+func TestHandleBatchRejectsOversizedBatch(t *testing.T) {
+	b := tmpKnoxDB(t)
+	node := &Node{Kdb: b, Name: "B"}
+	sv := httptest.NewServer(node.Handler())
+	defer sv.Close()
+
+	rows := make([]db.GossipObservation, maxBatchRows+1)
+	for i := range rows {
+		rows[i] = db.GossipObservation{
+			NodeID: "0123456789abcdef0123456789abcdef", HCL: int64(i + 1),
+			Fingerprint: "fp:oversized", SourceID: "test", Title: "t", Summary: "s",
+			CollectedAt: "2026-08-29T00:00:00Z",
+		}
+	}
+	body, _ := json.Marshal(rows)
+	resp, err := http.Post(sv.URL+"/v1/obs/batch", "application/json", bytes.NewReader(body))
+	if err != nil {
+		t.Fatalf("post: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusBadRequest {
+		t.Errorf("oversized batch: want 400, got %d", resp.StatusCode)
+	}
+}
+
+// TestHandleBatchRejectsHugeBody: an oversized body (beyond the 4 MiB cap)
+// must be refused even when the row count is small.
+func TestHandleBatchRejectsHugeBody(t *testing.T) {
+	b := tmpKnoxDB(t)
+	node := &Node{Kdb: b, Name: "B"}
+	sv := httptest.NewServer(node.Handler())
+	defer sv.Close()
+
+	rows := []db.GossipObservation{{
+		NodeID: "0123456789abcdef0123456789abcdef", HCL: 1,
+		Fingerprint: "fp:huge", SourceID: "test", Title: "t",
+		Summary: strings.Repeat("x", 5<<20), // 5 MiB summary
+		CollectedAt: "2026-08-29T00:00:00Z",
+	}}
+	body, _ := json.Marshal(rows)
+	resp, err := http.Post(sv.URL+"/v1/obs/batch", "application/json", bytes.NewReader(body))
+	if err != nil {
+		t.Fatalf("post: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusBadRequest {
+		t.Errorf("huge body: want 400, got %d", resp.StatusCode)
+	}
+}
+
+// TestHandleBatchSkipsSelfRows: rows claiming the receiver's own node_id are
+// dropped at the HTTP layer too (the poisoning vector), reported as conflicts.
+func TestHandleBatchSkipsSelfRows(t *testing.T) {
+	b := tmpKnoxDB(t)
+	node := &Node{Kdb: b, Name: "B"}
+	sv := httptest.NewServer(node.Handler())
+	defer sv.Close()
+
+	rows := []db.GossipObservation{
+		{NodeID: b.NodeID(), HCL: 1, Fingerprint: "fp:self1", SourceID: "test", Title: "t", Summary: "s", CollectedAt: "2026-08-29T00:00:00Z"},
+		{NodeID: b.NodeID(), HCL: 2, Fingerprint: "fp:self2", SourceID: "test", Title: "t", Summary: "s", CollectedAt: "2026-08-29T00:00:00Z"},
+	}
+	body, _ := json.Marshal(rows)
+	resp, err := http.Post(sv.URL+"/v1/obs/batch", "application/json", bytes.NewReader(body))
+	if err != nil {
+		t.Fatalf("post: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("want 200, got %d", resp.StatusCode)
+	}
+	var out struct {
+		Accepted int `json:"accepted"`
+		Conflict int `json:"conflict"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+		t.Fatalf("decode: %v", err)
+	}
+	if out.Accepted != 0 || out.Conflict != 2 {
+		t.Errorf("self rows: want accepted=0 conflict=2, got accepted=%d conflict=%d", out.Accepted, out.Conflict)
+	}
+}

internal/watch/reconcile.go

@@ -20,4 +20,4 @@ func Reconcile(kdb *db.KnoxDB) (int, int, error) {
 		return 0, 0, fmt.Errorf("auto-thread: %w", err)
 	}
 	return created, linked, nil
-}
+}

internal/watch/watch.go

@@ -1,15 +1,19 @@
 package watch
 
 import (
+	"io/fs"
 	"log"
 	"net/http"
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/david/knox/internal/db"
 	"github.com/david/knox/internal/ingest"
+	"github.com/david/knox/internal/metrics"
 	"github.com/fsnotify/fsnotify"
 )
 
@@ -24,6 +28,45 @@ type Watcher struct {
 	vault         string
 	debounce      time.Duration
 	fileIngesters []ingest.Ingester
+	metrics       *metrics.Metrics
+}
+
+// fileDebouncer schedules one ingest per path a settle-window after the last
+// relevant event (trailing edge): writers that emit bursts (multi-write appends,
+// atomic save = temp-write + rename) settle before anything is read, so a
+// partial file is never recorded as final state. Timers are removed once they
+// fire, keeping the map bounded by recently-active paths.
+type fileDebouncer struct {
+	mu     sync.Mutex
+	timers map[string]*time.Timer
+}
+
+func newFileDebouncer() *fileDebouncer {
+	return &fileDebouncer{timers: make(map[string]*time.Timer)}
+}
+
+func (d *fileDebouncer) schedule(path string, delay time.Duration, fn func(string)) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	if t, ok := d.timers[path]; ok {
+		t.Stop()
+	}
+	d.timers[path] = time.AfterFunc(delay, func() {
+		d.mu.Lock()
+		delete(d.timers, path)
+		d.mu.Unlock()
+		fn(path)
+	})
+}
+
+// cancel drops any pending ingest for path (e.g. the file was deleted).
+func (d *fileDebouncer) cancel(path string) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	if t, ok := d.timers[path]; ok {
+		t.Stop()
+		delete(d.timers, path)
+	}
 }
 
 func New(kdb *db.KnoxDB, dirs []string) *Watcher {
@@ -40,6 +83,7 @@ func New(kdb *db.KnoxDB, dirs []string) *Watcher {
 			ingest.NewLogIngester(),
 			ingest.NewSkillsIngester(),
 		},
+		metrics: metrics.New(kdb, "knox"),
 	}
 }
 
@@ -53,7 +97,7 @@ func (w *Watcher) Start() error {
 	// Start the gossip server first so peers can reach us while the initial
 	// seed is still ingesting. Vault/dirs are logged after the seed below.
 	gossipAddr := ListenAddr()
-	node := &Node{Kdb: w.knoxDB, Name: "knox", Addr: gossipAddr}
+	node := &Node{Kdb: w.knoxDB, Name: "knox", Addr: gossipAddr, Metrics: w.metrics}
 	srv := &http.Server{Addr: gossipAddr, Handler: node.Handler()}
 	go func() {
 		log.Printf("[knox] gossip listening on %s", gossipAddr)
@@ -61,6 +105,17 @@ func (w *Watcher) Start() error {
 			log.Printf("[knox] gossip server: %v", err)
 		}
 	}()
+
+	// Prometheus scraping on a dedicated port (KNOX_METRICS_ADDR).
+	metricsAddr := MetricsAddr()
+	metricsSrv := &http.Server{Addr: metricsAddr, Handler: node.Metrics.Handler()}
+	go func() {
+		log.Printf("[knox] metrics listening on %s", metricsAddr)
+		if err := metricsSrv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+			log.Printf("[knox] metrics server: %v", err)
+		}
+	}()
+
 	if peers := PeerAddrs(); len(peers) > 0 {
 		log.Printf("[knox] gossip peers: %v", peers)
 	}
@@ -72,17 +127,13 @@ func (w *Watcher) Start() error {
 		log.Printf("[knox] obsidian vault: %s", w.vault)
 	}
 
-	debounceMap := make(map[string]time.Time)
+	debounce := newFileDebouncer()
 	browserTicker := time.NewTicker(browserInterval)
 	giteaTicker := time.NewTicker(10 * time.Minute)
 	gitTicker := time.NewTicker(10 * time.Minute)
 	threadTicker := time.NewTicker(10 * time.Minute)
 	gossipTicker := time.NewTicker(gossipInterval)
-	browserRunning := false
-	giteaRunning := false
-	gitRunning := false
-	threadRunning := false
-	gossipRunning := false
+	var browserRunning, giteaRunning, gitRunning, threadRunning, gossipRunning atomic.Bool
 
 	for {
 		select {
@@ -90,63 +141,76 @@ func (w *Watcher) Start() error {
 			if !ok {
 				return nil
 			}
-			if !w.isRelevantEvent(event) {
-				continue
+
+			// fsnotify is non-recursive: files appearing inside newly created
+			// subdirectories would otherwise be invisible to the daemon.
+			isDir := false
+			if event.Has(fsnotify.Create) {
+				if info, err := os.Stat(event.Name); err == nil && info.IsDir() {
+					isDir = true
+					if err := watcher.Add(event.Name); err != nil {
+						log.Printf("[knox] cannot watch new dir %s: %v", event.Name, err)
+					} else {
+						log.Printf("[knox] watching new dir %s", event.Name)
+					}
+				}
 			}
 
-			now := time.Now()
-			if last, ok := debounceMap[event.Name]; ok && now.Sub(last) < w.debounce {
+			if event.Has(fsnotify.Remove) {
+				// Drop any pending ingest for a deleted file. (No tombstone is
+				// written yet — the entry lingers until reconcile/prune.)
+				debounce.cancel(event.Name)
+				continue
+			}
+			if isDir || !w.isRelevantEvent(event) {
 				continue
 			}
-			debounceMap[event.Name] = now
 
 			trigger := eventOpName(event.Op)
 			log.Printf("[knox] %s %s", trigger, filepath.Base(event.Name))
-			w.ingestFile(event.Name, trigger)
+			name := event.Name
+			debounce.schedule(name, w.debounce, func(path string) {
+				w.ingestFile(path, trigger)
+			})
 
 		case <-browserTicker.C:
-			if browserRunning {
+			if !browserRunning.CompareAndSwap(false, true) {
 				continue
 			}
-			browserRunning = true
 			go func() {
-				defer func() { browserRunning = false }()
+				defer browserRunning.Store(false)
 				w.ingestBrowserHistory()
 			}()
 		case <-giteaTicker.C:
-			if giteaRunning {
+			if !giteaRunning.CompareAndSwap(false, true) {
 				continue
 			}
-			giteaRunning = true
 			go func() {
-				defer func() { giteaRunning = false }()
+				defer giteaRunning.Store(false)
 				w.ingestGitea()
 			}()
 		case <-gitTicker.C:
-			if gitRunning {
+			if !gitRunning.CompareAndSwap(false, true) {
 				continue
 			}
-			gitRunning = true
 			go func() {
-				defer func() { gitRunning = false }()
+				defer gitRunning.Store(false)
 				w.ingestGit()
 			}()
 		case <-threadTicker.C:
-			if threadRunning {
+			if !threadRunning.CompareAndSwap(false, true) {
 				continue
 			}
-			threadRunning = true
 			go func() {
-				defer func() { threadRunning = false }()
+				defer threadRunning.Store(false)
 				w.autoThread()
 			}()
 		case <-gossipTicker.C:
-			if gossipRunning {
+			if !gossipRunning.CompareAndSwap(false, true) {
 				continue
 			}
-			gossipRunning = true
 			go func() {
-				defer func() { gossipRunning = false }()
+				defer gossipRunning.Store(false)
 				w.syncGossip()
 			}()
 
@@ -161,44 +225,38 @@ func (w *Watcher) Start() error {
 
 func (w *Watcher) seed(watcher *fsnotify.Watcher) {
 	for _, dir := range w.dirs {
-		abs, _ := filepath.Abs(dir)
-		if err := watcher.Add(abs); err != nil {
-			log.Printf("[knox] cannot watch %s: %v", abs, err)
-			continue
+		if err := w.watchTree(watcher, dir); err != nil {
+			log.Printf("[knox] cannot watch %s: %v", dir, err)
 		}
-		log.Printf("[knox] watching %s", abs)
 	}
 
-	// Watch Obsidian vault
+	// Watch Obsidian vault (every subdirectory, non-recursively mirrored)
 	if w.vault != "" {
-		if err := watcher.Add(w.vault); err != nil {
+		if err := w.watchTree(watcher, w.vault); err != nil {
 			log.Printf("[knox] cannot watch obsidian vault %s: %v", w.vault, err)
-		} else {
-			log.Printf("[knox] watching %s (obsidian)", w.vault)
 		}
 	}
 
-	// Seed existing files
-	for _, ing := range w.fileIngesters {
-		for _, dir := range w.dirs {
-			patterns := []string{
-				filepath.Join(dir, "*"),
-				filepath.Join(dir, "*", "SKILL.md"),
+	// Seed existing files anywhere under the watched dirs. fsnotify watches the
+	// whole tree, so live events cover any depth; this walk covers startup so
+	// pre-existing nested files (e.g. skills at two+ levels) are indexed too.
+	for _, dir := range w.dirs {
+		_ = filepath.WalkDir(dir, func(path string, d fs.DirEntry, err error) error {
+			if err != nil || d.IsDir() {
+				return nil // skip unreadable entries; dirs are handled by the watch
 			}
-			for _, pattern := range patterns {
-				entries, _ := filepath.Glob(pattern)
-				for _, path := range entries {
-					if !MatchesIngester(path, ing.SourceID()) {
-						continue
-					}
+			for _, ing := range w.fileIngesters {
+				if MatchesIngester(path, ing.SourceID()) {
 					w.ingestFileWith(path, ing, "seed")
+					return nil
 				}
 			}
-		}
+			return nil
+		})
 	}
 
 	// Seed Obsidian notes via the full walk: skips dot-dirs (.trash, .obsidian)
-	// and covers all depths — glob patterns would match dot-dirs and miss depth >2.
+	// and covers all depths.
 	if w.vault != "" {
 		if notes, err := ingest.NewObsidianIngester(w.vault).IngestAll(); err == nil {
 			for _, r := range notes {
@@ -208,6 +266,39 @@ func (w *Watcher) seed(watcher *fsnotify.Watcher) {
 	}
 }
 
+// watchTree adds a directory and every non-hidden subdirectory to the watcher,
+// mirroring fsnotify's non-recursive API with an explicit walk. Hidden
+// directories (.git, .obsidian, .trash) are skipped so their churn doesn't burn
+// inotify watches.
+func (w *Watcher) watchTree(watcher *fsnotify.Watcher, root string) error {
+	rootAbs, err := filepath.Abs(root)
+	if err != nil {
+		return err
+	}
+	if err := watcher.Add(rootAbs); err != nil {
+		return err
+	}
+	log.Printf("[knox] watching %s", rootAbs)
+	return filepath.WalkDir(rootAbs, func(path string, d fs.DirEntry, err error) error {
+		if err != nil {
+			return nil
+		}
+		if !d.IsDir() {
+			return nil
+		}
+		if path != rootAbs && strings.HasPrefix(d.Name(), ".") {
+			return filepath.SkipDir
+		}
+		if path == rootAbs {
+			return nil
+		}
+		if err := watcher.Add(path); err != nil {
+			log.Printf("[knox] cannot watch %s: %v", path, err)
+		}
+		return nil
+	})
+}
+
 func (w *Watcher) ingestFile(path, trigger string) {
 	for _, ing := range w.fileIngesters {
 		if MatchesIngester(path, ing.SourceID()) {
@@ -442,29 +533,23 @@ func (w *Watcher) syncGossip() {
 		return
 	}
 
-	before := 0
-	if n, err := w.knoxDB.EntryCount(); err == nil {
-		before = n
-	}
+	pulled := Run(w.knoxDB, w.metrics, peers)
 
-	Run(w.knoxDB, peers)
-
-	// If new observations arrived, reconcile to pick up entries/threads they
-	// imply (deterministic log → derived rebuild).
-	if after, err := w.knoxDB.EntryCount(); err == nil && after > before {
+	// A pull only appends to the observation log; the entries cache and
+	// auto-threads are derived state that reconcile rebuilds. Comparing entry
+	// counts can never trigger this (pushes/pulls never touch entries directly),
+	// so reconcile fires on the sweep's newly-inserted observation count.
+	if pulled > 0 {
 		created, linked, err := Reconcile(w.knoxDB)
 		if err != nil {
 			log.Printf("[knox] gossip reconcile: %v", err)
 			return
 		}
-		log.Printf("[knox] gossip reconcile done: %d created, %d linked", created, linked)
+		log.Printf("[knox] gossip reconcile done: %d created, %d linked after pulling %d obs", created, linked, pulled)
 	}
 }
 
 func (w *Watcher) isRelevantEvent(event fsnotify.Event) bool {
-	if event.Has(fsnotify.Remove) || event.Has(fsnotify.Rename) {
-		return false
-	}
 	name := filepath.Base(event.Name)
 	// Opencode session diffs and logs
 	if strings.HasPrefix(name, "ses_") || strings.HasSuffix(name, ".log") {
@@ -498,6 +583,8 @@ func eventOpName(op fsnotify.Op) string {
 		return "inotify:WRITE"
 	case op.Has(fsnotify.Chmod):
 		return "inotify:CHMOD"
+	case op.Has(fsnotify.Rename):
+		return "inotify:RENAME"
 	default:
 		return "inotify:UNKNOWN"
 	}

main.go

@@ -5,11 +5,11 @@ import (
 	"log"
 	"os"
 
-	mcpServer "github.com/mark3labs/mcp-go/server"
-	"github.com/spf13/cobra"
 	knoxcmd "github.com/david/knox/internal/cmd"
 	"github.com/david/knox/internal/db"
 	knoxserver "github.com/david/knox/internal/server"
+	mcpServer "github.com/mark3labs/mcp-go/server"
+	"github.com/spf13/cobra"
 )
 
 func newServeCmd(kdb *db.KnoxDB) *cobra.Command {