diff --git a/internal/cmd/gossip.go b/internal/cmd/gossip.go index d0b18e1..eadca05 100644 --- a/internal/cmd/gossip.go +++ b/internal/cmd/gossip.go @@ -2,6 +2,7 @@ package cmd import ( "fmt" + "time" "github.com/david/knox/internal/db" "github.com/david/knox/internal/watch" @@ -77,7 +78,7 @@ func NewGossipCmd(kdb *db.KnoxDB) *cobra.Command { // gossipDiff compares this node's observation log + auto-threads with a peer's // via /v1/diff. Never writes; it is the preview for what a sync would adopt. func gossipDiff(kdb *db.KnoxDB, peerAddr string) error { - c := &watch.Client{Addr: peerAddr} + c := &watch.Client{Addr: peerAddr, Timeout: 10 * time.Second} remote, err := c.Diff() if err != nil { return err diff --git a/internal/cmd/ingest.go b/internal/cmd/ingest.go index cbb1039..de762a0 100644 --- a/internal/cmd/ingest.go +++ b/internal/cmd/ingest.go @@ -97,6 +97,8 @@ func NewIngestCmd(kdb *db.KnoxDB) *cobra.Command { for _, r := range results { record(r) } + } else { + log.Printf("[knox] obsidian ingest failed: %v", err) } } log.Printf("[knox] ingesting browser-history...") @@ -104,12 +106,16 @@ func NewIngestCmd(kdb *db.KnoxDB) *cobra.Command { for _, r := range results { record(r) } + } else { + log.Printf("[knox] browser-history ingest failed: %v", err) } log.Printf("[knox] ingesting gitea...") if results, err := ingest.NewGiteaIngester().IngestAll(); err == nil { for _, r := range results { record(r) } + } else { + log.Printf("[knox] gitea ingest failed: %v", err) } fmt.Printf("\nIngest complete: %d new entries, %d updated\n", totalNew, totalUpdated) diff --git a/internal/cmd/watch.go b/internal/cmd/watch.go index f9a3461..e9300c8 100644 --- a/internal/cmd/watch.go +++ b/internal/cmd/watch.go @@ -1,8 +1,10 @@ package cmd import ( + "fmt" "io" "log" + "os" "github.com/david/knox/internal/db" "github.com/david/knox/internal/watch" @@ -25,7 +27,10 @@ func NewWatchCmd(kdb *db.KnoxDB) *cobra.Command { log.Printf("[knox] starting watcher, scanning %d directories", len(dirs)) w := watch.New(kdb, dirs) if err := w.Start(); err != nil { - log.Fatalf("watcher error: %v", err) + // Not log.Fatalf: --quiet redirects the logger to io.Discard, + // so a fatal exit must reach the user on stderr directly. + fmt.Fprintf(os.Stderr, "watcher error: %v\n", err) + os.Exit(1) } }, } diff --git a/internal/db/db.go b/internal/db/db.go index 4b8d453..d615dc4 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -308,18 +308,21 @@ func (k *KnoxDB) RecordObservation(o ObservationRecord) (obsID int64, isNew bool `INSERT OR IGNORE INTO thread_observations (thread_id, observation_id, relevance) VALUES (?, ?, 'auto')`, goldenID, obsID, ) - linked := err == nil && res != nil - if linked { - if n, _ := res.RowsAffected(); n == 0 { - linked = false - } + if err != nil { + return 0, false, fmt.Errorf("auto-link golden thread: %w", err) + } + affected, _ := res.RowsAffected() + linked := affected > 0 + if _, err := tx.Exec(`UPDATE threads SET updated_at=datetime('now') WHERE id=?`, goldenID); err != nil { + return 0, false, fmt.Errorf("bump golden thread updated_at: %w", err) } - tx.Exec(`UPDATE threads SET updated_at=datetime('now') WHERE id=?`, goldenID) // Temporal cross-reference: when a high-signal non-browser event links, // nearby browser history (±1h) becomes thread context too. if linked && o.SourceID != "browser-history" && o.CreatedAt != "" { - k.linkTemporalNeighbors(tx, goldenID, o.CreatedAt, keywords, bigrams) + if err := k.linkTemporalNeighbors(tx, goldenID, o.CreatedAt, keywords, bigrams); err != nil { + return 0, false, fmt.Errorf("link temporal neighbors: %w", err) + } } } } @@ -330,7 +333,7 @@ func (k *KnoxDB) RecordObservation(o ObservationRecord) (obsID int64, isNew bool // linkTemporalNeighbors links browser-history observations within ±1h of a // signal timestamp to the thread. A weak keyword check still applies so // unrelated browsing (lunch news) doesn't pollute the thread. -func (k *KnoxDB) linkTemporalNeighbors(tx *sql.Tx, threadID int64, createdAt string, keywords []string, bigrams [][2]string) { +func (k *KnoxDB) linkTemporalNeighbors(tx *sql.Tx, threadID int64, createdAt string, keywords []string, bigrams [][2]string) error { rows, err := tx.Query( `SELECT o.id, COALESCE(o.title,''), COALESCE(o.summary,'') FROM observations o @@ -341,7 +344,7 @@ func (k *KnoxDB) linkTemporalNeighbors(tx *sql.Tx, threadID int64, createdAt str createdAt, ) if err != nil { - return + return err } defer rows.Close() @@ -349,15 +352,18 @@ func (k *KnoxDB) linkTemporalNeighbors(tx *sql.Tx, threadID int64, createdAt str var obsID int64 var title, summary string if err := rows.Scan(&obsID, &title, &summary); err != nil { - continue + return err } if isRelevant(title, summary, "", keywords, bigrams) || weakRelevant(title, summary, keywords) { - tx.Exec( + if _, err := tx.Exec( `INSERT OR IGNORE INTO thread_observations (thread_id, observation_id, relevance) VALUES (?, ?, 'temporal')`, threadID, obsID, - ) + ); err != nil { + return err + } } } + return rows.Err() } // weakRelevant is a lower bar for temporal neighbors: any single keyword match. @@ -447,23 +453,39 @@ func (k *KnoxDB) RebuildEntriesFromObservations() (before, after int, err error) if _, err := tx.Exec("DELETE FROM entries"); err != nil { return 0, 0, err } + // Content fields come from the latest observation per fingerprint, folded in + // a total order (hcl DESC, node_id DESC, id DESC) — deterministic for any two + // nodes holding identical logs, per the gossip spec (bare GROUP BY columns + // would take an arbitrary row, merge-order dependent). _, 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) SELECT - o.fingerprint, - o.source_id, - o.source_path, - o.project, - o.content_type, - o.title, - o.summary, - MIN(o.collected_at), - MAX(o.collected_at), - COALESCE(NULLIF(MAX(o.created_at),''), MIN(o.collected_at)), - COUNT(*), - MAX(o.confidence) - FROM observations o - GROUP BY o.fingerprint`, + latest.fingerprint, + latest.source_id, + latest.source_path, + latest.project, + latest.content_type, + latest.title, + latest.summary, + agg.first_seen, + agg.last_seen, + agg.created_at, + agg.ref_count, + agg.max_confidence + FROM ( + SELECT *, ROW_NUMBER() OVER (PARTITION BY fingerprint ORDER BY hcl DESC, node_id DESC, id DESC) AS rn + FROM observations + ) latest + JOIN ( + SELECT fingerprint, + MIN(collected_at) AS first_seen, + MAX(collected_at) AS last_seen, + COALESCE(NULLIF(MAX(created_at),''), MIN(collected_at)) AS created_at, + COUNT(*) AS ref_count, + MAX(confidence) AS max_confidence + FROM observations GROUP BY fingerprint + ) agg ON agg.fingerprint = latest.fingerprint + WHERE latest.rn = 1`, ) if err != nil { return 0, 0, fmt.Errorf("rebuild entries: %w", err) @@ -604,6 +626,9 @@ func scanEntries(rows *sql.Rows) ([]Entry, error) { } entries = append(entries, e) } + if err := rows.Err(); err != nil { + return nil, err + } return entries, nil } @@ -655,38 +680,50 @@ func (k *KnoxDB) MarkSessionIndexed(sessionID string) error { func (k *KnoxDB) Stats() (map[string]any, error) { stats := make(map[string]any) - var v int - var s string + // A broken DB reports an error, not zeros: every aggregate is checked. + countInt := func(query string) (int, error) { + var v int + if err := k.db.QueryRow(query).Scan(&v); err != nil { + return 0, err + } + return v, nil + } + ints := []struct { + key string + query string + }{ + {"total_entries", "SELECT COUNT(*) FROM entries"}, + {"entries_last_24h", "SELECT COUNT(*) FROM entries WHERE last_seen > datetime('now', '-1 day')"}, + {"total_projects", "SELECT COUNT(DISTINCT project) FROM entries"}, + {"total_observations", "SELECT COUNT(*) FROM observations"}, + {"observations_last_24h", "SELECT COUNT(*) FROM observations WHERE collected_at > datetime('now', '-1 day')"}, + {"total_sessions", "SELECT COUNT(*) FROM sessions"}, + {"pending_reflections", "SELECT COUNT(*) FROM sessions WHERE indexed=0"}, + } + for _, it := range ints { + v, err := countInt(it.query) + if err != nil { + return nil, fmt.Errorf("stats %s: %w", it.key, err) + } + stats[it.key] = v + } - k.db.QueryRow("SELECT COUNT(*) FROM entries").Scan(&v) - stats["total_entries"] = v - v = 0 - k.db.QueryRow("SELECT COUNT(*) FROM entries WHERE last_seen > datetime('now', '-1 day')").Scan(&v) - stats["entries_last_24h"] = v - v = 0 - k.db.QueryRow("SELECT COUNT(DISTINCT project) FROM entries").Scan(&v) - stats["total_projects"] = v - v = 0 - k.db.QueryRow("SELECT COUNT(*) FROM observations").Scan(&v) - stats["total_observations"] = v - v = 0 - k.db.QueryRow("SELECT COUNT(*) FROM observations WHERE collected_at > datetime('now', '-1 day')").Scan(&v) - stats["observations_last_24h"] = v - v = 0 - k.db.QueryRow("SELECT COALESCE(MIN(collected_at),'') FROM observations").Scan(&s) - stats["earliest_observation"] = s - s = "" - k.db.QueryRow("SELECT COUNT(*) FROM sessions").Scan(&v) - stats["total_sessions"] = v - v = 0 - k.db.QueryRow("SELECT COUNT(*) FROM sessions WHERE indexed=0").Scan(&v) - stats["pending_reflections"] = v - v = 0 + var earliest string + if err := k.db.QueryRow("SELECT COALESCE(MIN(collected_at),'') FROM observations").Scan(&earliest); err != nil { + return nil, fmt.Errorf("stats earliest_observation: %w", err) + } + stats["earliest_observation"] = earliest - goldenID, _ := k.GoldenThreadID() + goldenID, err := k.GoldenThreadID() + if err != nil { + return nil, fmt.Errorf("stats golden_thread: %w", err) + } if goldenID > 0 { stats["golden_thread_id"] = goldenID - t, _ := k.GetThread(goldenID) + t, err := k.GetThread(goldenID) + if err != nil { + return nil, fmt.Errorf("stats golden_thread: %w", err) + } if t != nil { stats["golden_thread"] = t.Title } @@ -989,20 +1026,26 @@ func (k *KnoxDB) CreateThread(title, motivation, priority, tags, provenance stri } 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 - } - } + // INSERT OR IGNORE makes creation atomic under the partial unique index: + // concurrent creators (thread ticker + post-pull reconcile, or two gossip + // nodes converging on the same cluster) lose the race into a no-op instead + // of a UNIQUE error, then pick up the winner's id. res, err := k.db.Exec( - `INSERT INTO threads (title, motivation, priority, tags, provenance, cluster_key) VALUES (?, ?, ?, ?, ?, ?)`, + `INSERT OR IGNORE INTO threads (title, motivation, priority, tags, provenance, cluster_key) VALUES (?, ?, ?, ?, ?, ?)`, title, motivation, priority, tags, provenance, clusterKey, ) if err != nil { return 0, false, err } - id, _ := res.LastInsertId() - return id, true, nil + if n, _ := res.RowsAffected(); n > 0 { + id, _ := res.LastInsertId() + return id, true, nil + } + // Ignored insert: a thread with this cluster key already exists. + if existing := k.ThreadByClusterKey(clusterKey); existing != 0 { + return existing, false, nil + } + return 0, false, fmt.Errorf("insert ignored but no thread with cluster key %q", clusterKey) } // ThreadByClusterKey returns the id of the thread auto-created for a given @@ -1107,15 +1150,21 @@ func (k *KnoxDB) UpdateThread(id int64, title, motivation, priority, tags string } func (k *KnoxDB) LinkObservationToThread(threadID, observationID int64, relevance string) error { - _, err := k.db.Exec( - `INSERT OR IGNORE INTO thread_observations (thread_id, observation_id, relevance) VALUES (?, ?, ?)`, - threadID, observationID, relevance, - ) + tx, err := k.db.Begin() if err != nil { return err } - _, err = k.db.Exec(`UPDATE threads SET updated_at=datetime('now') WHERE id=?`, threadID) - return err + defer tx.Rollback() + if _, err := tx.Exec( + `INSERT OR IGNORE INTO thread_observations (thread_id, observation_id, relevance) VALUES (?, ?, ?)`, + threadID, observationID, relevance, + ); err != nil { + return err + } + if _, err := tx.Exec(`UPDATE threads SET updated_at=datetime('now') WHERE id=?`, threadID); err != nil { + return err + } + return tx.Commit() } // AutoLinkThreadObservations resolves each entry fingerprint to its most recent @@ -1130,6 +1179,7 @@ func (k *KnoxDB) AutoLinkThreadObservations(threadID int64, fingerprints []strin defer tx.Rollback() linked := 0 + var firstErr error for _, fp := range fingerprints { if fp == "" { continue @@ -1138,6 +1188,11 @@ func (k *KnoxDB) AutoLinkThreadObservations(threadID int64, fingerprints []strin if err := tx.QueryRow( `SELECT id FROM observations WHERE fingerprint=? ORDER BY hcl DESC, id DESC LIMIT 1`, fp, ).Scan(&obsID); err != nil { + // A fingerprint without observations is expected (provenance links); + // a real query failure is not — remember the first one. + if err != sql.ErrNoRows && firstErr == nil { + firstErr = err + } continue } res, err := tx.Exec( @@ -1145,6 +1200,9 @@ func (k *KnoxDB) AutoLinkThreadObservations(threadID int64, fingerprints []strin threadID, obsID, ) if err != nil { + if firstErr == nil { + firstErr = err + } continue } if n, _ := res.RowsAffected(); n > 0 { @@ -1157,7 +1215,10 @@ func (k *KnoxDB) AutoLinkThreadObservations(threadID int64, fingerprints []strin return linked, err } } - return linked, tx.Commit() + if err := tx.Commit(); err != nil { + return linked, err + } + return linked, firstErr } // ActiveThreadByKeyword returns the most recently updated active thread whose @@ -1226,15 +1287,26 @@ func (k *KnoxDB) ThreadObservations(threadID int64) ([]Observation, error) { } func (k *KnoxDB) AddThreadNote(threadID int64, note string) (int64, error) { - res, err := k.db.Exec( + tx, err := k.db.Begin() + if err != nil { + return 0, err + } + defer tx.Rollback() + res, err := tx.Exec( `INSERT INTO thread_notes (thread_id, note) VALUES (?, ?)`, threadID, note, ) if err != nil { return 0, err } - _, err = k.db.Exec(`UPDATE threads SET updated_at=datetime('now') WHERE id=?`, threadID) - return res.LastInsertId() + if _, err := tx.Exec(`UPDATE threads SET updated_at=datetime('now') WHERE id=?`, threadID); err != nil { + return 0, err + } + id, err := res.LastInsertId() + if err != nil { + return 0, err + } + return id, tx.Commit() } func (k *KnoxDB) ThreadNotes(threadID int64) ([]struct { diff --git a/internal/db/db_test.go b/internal/db/db_test.go index f8d326f..4446c5c 100644 --- a/internal/db/db_test.go +++ b/internal/db/db_test.go @@ -154,3 +154,89 @@ func TestOpenReopenHCLMonotonicAcrossRestart(t *testing.T) { t.Errorf("unexpected row above max: %s", after[0].Fingerprint) } } + +// TestCreateThreadClusterConcurrentIdempotent: two concurrent creators of the +// same cluster key must converge on one thread (one created=true, one false) +// with no UNIQUE constraint error — the INSERT OR IGNORE path. +func TestCreateThreadClusterConcurrentIdempotent(t *testing.T) { + k := tmpDB(t) + const key = "cluster:race" + var wg sync.WaitGroup + ids := make([]int64, 2) + created := make([]bool, 2) + errs := make([]error, 2) + for i := 0; i < 2; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + ids[i], created[i], errs[i] = k.CreateThreadCluster("Race thread", "m", "medium", "[]", `{}`, key) + }(i) + } + wg.Wait() + for i := range errs { + if errs[i] != nil { + t.Fatalf("creator %d: %v", i, errs[i]) + } + } + if ids[0] != ids[1] { + t.Errorf("concurrent creators got different ids: %d vs %d", ids[0], ids[1]) + } + if created[0] == created[1] { + t.Errorf("exactly one creator should report created=true, got %v %v", created[0], created[1]) + } + threads, err := k.ListThreads("") + if err != nil { + t.Fatal(err) + } + n := 0 + for _, th := range threads { + if th.Title == "Race thread" { + n++ + } + } + if n != 1 { + t.Errorf("want exactly 1 race thread, got %d", n) + } +} + +// TestRebuildEntriesDeterministicFold: two nodes holding the same observations +// in different merge orders must rebuild identical derived state — content +// fields come from the highest-HCL observation per fingerprint (spec 6.2), +// never from an arbitrary (merge-order-dependent) row. +func TestRebuildEntriesDeterministicFold(t *testing.T) { + const foreign = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + rows := []GossipObservation{ + {NodeID: foreign, HCL: 10, Fingerprint: "fp:fold", SourceID: "test", Title: "old title", Summary: "old", CollectedAt: "2026-08-29T00:00:00Z", Confidence: 0.5}, + {NodeID: foreign, HCL: 20, Fingerprint: "fp:fold", SourceID: "test", Title: "new title", Summary: "new", CollectedAt: "2026-08-29T01:00:00Z", Confidence: 0.9}, + } + results := make([]Entry, 2) + for i, order := range [][]GossipObservation{{rows[0], rows[1]}, {rows[1], rows[0]}} { + k := tmpDB(t) + if _, err := k.PushObservations(order); err != nil { + t.Fatalf("push: %v", err) + } + if _, _, err := k.RebuildEntriesFromObservations(); err != nil { + t.Fatalf("rebuild: %v", err) + } + e, err := k.FindEntry("fp:fold") + if err != nil || e == nil { + t.Fatalf("find entry: %v", err) + } + results[i] = *e + } + if results[0].Title != "new title" { + t.Errorf("order [old,new]: title = %q, want %q", results[0].Title, "new title") + } + if results[1].Title != "new title" { + t.Errorf("order [new,old]: title = %q, want %q", results[1].Title, "new title") + } + if results[0] != results[1] { + t.Errorf("derived state diverged between merge orders:\n%+v\n%+v", results[0], results[1]) + } + if results[0].RefCount != 2 { + t.Errorf("ref_count = %d, want 2", results[0].RefCount) + } + if results[0].FirstSeen != "2026-08-29T00:00:00Z" || results[0].LastSeen != "2026-08-29T01:00:00Z" { + t.Errorf("first/last seen = %q/%q", results[0].FirstSeen, results[0].LastSeen) + } +} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 341e8ac..a2bbb43 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -59,14 +59,14 @@ func New(kdb *db.KnoxDB, nodeName string) *Metrics { 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.observationsGauge = newGaugeVec(reg, "knox_observations", "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.entriesGauge = newGauge(reg, "knox_entries", "Materialized entry cache size.") + m.projectsGauge = newGauge(reg, "knox_projects", "Distinct projects in the entry cache.") + m.sessionsGauge = newGauge(reg, "knox_sessions", "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.peersGauge = newGauge(reg, "knox_peers", "Known peer nodes.") + m.threadsByStatus = newGaugeVec(reg, "knox_threads", "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") diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go index 10c1955..b4ab3e1 100644 --- a/internal/metrics/metrics_test.go +++ b/internal/metrics/metrics_test.go @@ -79,7 +79,7 @@ func TestMetricsScrape(t *testing.T) { out := string(body) for _, want := range []string{ `knox_node_info{name="testnode"`, - `knox_observations_total{source_id="git"} 2`, + `knox_observations{source_id="git"} 2`, `knox_gossip_pulls_total 0`, "go_goroutines", "process_cpu_seconds_total", diff --git a/internal/server/server.go b/internal/server/server.go index cbaab96..5c7b93e 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -42,7 +42,8 @@ func New(kdb *db.KnoxDB) *Server { func (s *Server) Serve(addr string) error { log.Printf("[knox] web UI at http://%s", addr) - return http.ListenAndServe(addr, s.mux) + srv := &http.Server{Addr: addr, Handler: s.mux, ReadHeaderTimeout: 10 * time.Second, IdleTimeout: 60 * time.Second} + return srv.ListenAndServe() } // ─── Dashboard ─────────────────────────────────────────────── diff --git a/internal/watch/gossip.go b/internal/watch/gossip.go index c92310b..a6a6562 100644 --- a/internal/watch/gossip.go +++ b/internal/watch/gossip.go @@ -242,6 +242,9 @@ func (c *Client) Push(kdb *db.KnoxDB, peerVector map[string]int64) (int, error) return 0, err } defer resp.Body.Close() + if resp.StatusCode/100 != 2 { + return 0, fmt.Errorf("push to %s: %s", c.Addr, resp.Status) + } var out struct { Accepted int `json:"accepted"` } @@ -354,7 +357,13 @@ func Run(kdb *db.KnoxDB, m *metrics.Metrics, static []string) int { m.IncrementPull(pulled) } - pushed, _ := c.Push(kdb, p.Vector) + pushed, err := c.Push(kdb, p.Vector) + if err != nil { + log.Printf("[gossip] push %s: %v", addr, err) + if m != nil { + m.IncrementErrors() + } + } if m != nil { m.IncrementPush(pushed) } diff --git a/internal/watch/gossip_test.go b/internal/watch/gossip_test.go index 2e8dca6..11c40d6 100644 --- a/internal/watch/gossip_test.go +++ b/internal/watch/gossip_test.go @@ -8,6 +8,7 @@ import ( "path/filepath" "strings" "testing" + "time" "github.com/david/knox/internal/db" ) @@ -299,7 +300,7 @@ func TestHandleBatchRejectsHugeBody(t *testing.T) { rows := []db.GossipObservation{{ NodeID: "0123456789abcdef0123456789abcdef", HCL: 1, Fingerprint: "fp:huge", SourceID: "test", Title: "t", - Summary: strings.Repeat("x", 5<<20), // 5 MiB summary + Summary: strings.Repeat("x", 5<<20), // 5 MiB summary CollectedAt: "2026-08-29T00:00:00Z", }} body, _ := json.Marshal(rows) @@ -345,3 +346,20 @@ func TestHandleBatchSkipsSelfRows(t *testing.T) { t.Errorf("self rows: want accepted=0 conflict=2, got accepted=%d conflict=%d", out.Accepted, out.Conflict) } } + +// TestClientPushSurfacesHTTPError: a non-2xx push response must surface as an +// error — silently treating it as accepted=0 would hide a broken sync direction. +func TestClientPushSurfacesHTTPError(t *testing.T) { + a := tmpKnoxDB(t) + seedObs(a, "AAA") + + sv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "boom", http.StatusInternalServerError) + })) + defer sv.Close() + + c := &Client{Addr: sv.URL, Timeout: 5 * time.Second} + if _, err := c.Push(a, map[string]int64{}); err == nil { + t.Fatal("expected push error on HTTP 500, got nil") + } +} diff --git a/internal/watch/watch.go b/internal/watch/watch.go index 36b923d..4d0fda5 100644 --- a/internal/watch/watch.go +++ b/internal/watch/watch.go @@ -98,7 +98,7 @@ func (w *Watcher) Start() error { // seed is still ingesting. Vault/dirs are logged after the seed below. gossipAddr := ListenAddr() node := &Node{Kdb: w.knoxDB, Name: "knox", Addr: gossipAddr, Metrics: w.metrics} - srv := &http.Server{Addr: gossipAddr, Handler: node.Handler()} + srv := &http.Server{Addr: gossipAddr, Handler: node.Handler(), ReadHeaderTimeout: 10 * time.Second, IdleTimeout: 60 * time.Second} go func() { log.Printf("[knox] gossip listening on %s", gossipAddr) if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { @@ -108,7 +108,7 @@ func (w *Watcher) Start() error { // Prometheus scraping on a dedicated port (KNOX_METRICS_ADDR). metricsAddr := MetricsAddr() - metricsSrv := &http.Server{Addr: metricsAddr, Handler: node.Metrics.Handler()} + metricsSrv := &http.Server{Addr: metricsAddr, Handler: node.Metrics.Handler(), ReadHeaderTimeout: 10 * time.Second, IdleTimeout: 60 * time.Second} go func() { log.Printf("[knox] metrics listening on %s", metricsAddr) if err := metricsSrv.ListenAndServe(); err != nil && err != http.ErrServerClosed { @@ -367,8 +367,9 @@ func (w *Watcher) recordResult(result *ingest.IngestResult, trigger string) { if result.SourceID == "opencode-session" { sessionID, _ := result.Provenance["session_id"].(string) if sessionID != "" { - status := "active" - w.knoxDB.UpsertSession(sessionID, result.Project, result.Title, status) + if err := w.knoxDB.UpsertSession(sessionID, result.Project, result.Title, "active"); err != nil { + log.Printf("[knox] session upsert error: %v", err) + } } } } diff --git a/main.go b/main.go index 6cfa0e8..97cf271 100644 --- a/main.go +++ b/main.go @@ -1,6 +1,8 @@ package main import ( + "context" + "errors" "fmt" "log" "os" @@ -75,6 +77,11 @@ and maintains a searchable index. Use 'knox watch' for daemon mode.`, s := knoxcmd.NewMCPServer(kdb) log.Printf("[knox] starting MCP stdio server") if err := mcpServer.ServeStdio(s); err != nil { + // SIGINT/SIGTERM cancels the server context: clean shutdown, + // not a failure. + if errors.Is(err, context.Canceled) { + return nil + } return err } return nil