nudgebee · blue4209211 · Mar 10, 2026 · Mar 10, 2026
diff --git a/parser_test.go b/parser_test.go
@@ -13,10 +13,10 @@ func TestParser(t *testing.T) {
 	ch := make(chan LogEntry)
 	parser := NewParser(ch, nil, nil, time.Second, 256, false)
 
-	ch <- LogEntry{Timestamp: time.Now(), Content: "INFO:root:AWS access key: Key", Level: LevelInfo}
+	ch <- LogEntry{Timestamp: time.Now(), Content: "INFO:root:AWS access key: AKIAIOSFODNN7EXAMPLE", Level: LevelInfo}
 
-	// wait for 10 seconds
-	time.Sleep(10 * time.Second)
+	// Wait for multiline collector flush (1s timeout) + processing.
+	time.Sleep(3 * time.Second)
 	counts := parser.GetSensitiveCounters()
 	assert.Equal(t, 1, len(counts))
 	parser.Stop()

diff --git a/pattern.go b/pattern.go
@@ -212,23 +212,57 @@ func removeQuotedAndBrackets(s string, buf *bytes.Buffer) string {
 	return buf.String()
 }
 
+// jsonMessageKeys lists the JSON field names (lowercase) used for pattern extraction.
+// Following industry standard (Datadog, New Relic, Elastic, Better Stack), pattern
+// hashing uses only the message/error content, not metadata fields like timestamps,
+// file paths, line numbers, IDs, or data blobs which produce unstable hashes.
+var jsonMessageKeys = []string{"msg", "message", "error", "err", "reason", "log", "text"}
+
+// maxFallbackFieldLen caps individual field values in the fallback path to prevent
+// large data blobs (HTML, XML, stack traces) from overwhelming the pattern.
+const maxFallbackFieldLen = 200
+
 func normalizeJSONLog(line string) string {
 	var m map[string]interface{}
 	if err := json.Unmarshal([]byte(line), &m); err != nil {
-		// not a JSON log, return raw
 		return line
 	}
 
+	// Build a lowercase-key lookup for case-insensitive matching.
+	lowerMap := make(map[string]interface{}, len(m))
+	for k, v := range m {
+		lowerMap[strings.ToLower(k)] = v
+	}
+
+	// Extract only message-relevant fields for stable pattern hashing.
+	var buf strings.Builder
+	for _, k := range jsonMessageKeys {
+		if v, ok := lowerMap[k]; ok {
+			s := fmt.Sprintf("%v", v)
+			if s != "" {
+				buf.WriteString(s)
+				buf.WriteByte(' ')
+			}
+		}
+	}
+	if buf.Len() > 0 {
+		return strings.TrimSpace(buf.String())
+	}
+
+	// Fallback: no known message fields found, use all values sorted by key
+	// with a per-field length cap to prevent data blobs from dominating.
 	var keys []string
 	for k := range m {
 		keys = append(keys, k)
 	}
-	sort.Strings(keys) // sort keys alphabetically
-
-	var buf strings.Builder
+	sort.Strings(keys)
 	for _, k := range keys {
-		strVal := fmt.Sprintf("%v", m[k])
-		buf.WriteString(fmt.Sprintf("%s ", strVal))
+		s := fmt.Sprintf("%v", m[k])
+		if len(s) > maxFallbackFieldLen {
+			s = s[:maxFallbackFieldLen]
+		}
+		buf.WriteString(s)
+		buf.WriteByte(' ')
 	}
-	return strings.TrimSpace(buf.String()) // trim trailing space
+	return strings.TrimSpace(buf.String())
 }
diff --git a/pattern_test.go b/pattern_test.go
@@ -2,6 +2,9 @@ package logparser
 
 import (
 	"bytes"
+	"encoding/json"
+	"os"
+	"sort"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -116,12 +119,251 @@ func TestPatternRemoveQuotedAndBrackets(t *testing.T) {
 }
 
 func TestJsonPattern(t *testing.T) {
+	// Fallback path: no known message keys, uses all values.
 	p1 := NewPattern("{\"foo\": \"bar\"}")
 	p2 := NewPattern("{\"foo\": \"bar 11818181\"}")
 	assert.Equal(t, p1.String(), p2.String())
 	assert.Equal(t, p1.Hash(), p2.Hash())
 
+	// Python structured log with "message" key: pattern comes from message field only.
 	p3 := NewPattern("{\"asctime\": \"2025-04-03 08:10:06,482\", \"levelname\": \"ERROR\", \"filename\": \"database.py\", \"lineno\": 61, \"message\": \"Error selecting data from cloud_accounts table: connection to server at \\\"192.168.1.13\\\", port 5432 failed: FATAL: remaining connection slots are reserved for roles with the SUPERUSER attribute\\n\", \"exc_info\": \"Traceback (most recent call last):\\n File \\\"/app/db/database.py\\\", line 40, in select_data\\n conn = create_db_connection_pool().getconn()\\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\\n File \\\"/usr/local/lib/python3.12/site-packages/psycopg2/pool.py\\\", line 93, in _getconn\\n return self._connect(key)\\n ^^^^^^^^^^^^^^^^^^\\n File \\\"/usr/local/lib/python3.12/site-packages/psycopg2/pool.py\\\", line 63, in _connect\\n conn = psycopg2.connect(*self._args, **self._kwargs)\\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\\n File \\\"/usr/local/lib/python3.12/site-packages/psycopg2/__init__.py\\\", line 122, in connect\\n conn = _connect(dsn, connection_factory=connection_factory, **kwasync)\\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\\npsycopg2.OperationalError: connection to server at \\\"192.168.1.13\\\", port 5432 failed: FATAL: remaining connection slots are reserved for roles with the SUPERUSER attribute\\n\", \"taskName\": null}")
-
 	assert.NotEqual(t, "d4c3b7e0f4a1d2c5e8f9a4b5d6e7f8a9", p3.Hash())
+	// Pattern should come from the "message" field content, not exc_info/filename/etc.
+	assert.Contains(t, p3.String(), "Error")
+	assert.Contains(t, p3.String(), "selecting")
+	assert.NotContains(t, p3.String(), "database.py")
+	assert.NotContains(t, p3.String(), "Traceback")
+}
+
+func TestJsonPatternMessageFieldOnly(t *testing.T) {
+	emptyHash := NewPattern("").Hash()
+
+	// Same "msg" with different context fields must produce the same hash.
+	log1 := `{"time":"2026-03-10T05:02:40Z","level":"ERROR","msg":"search: no links found","file":"/app/tools/tool_web_search.go","line":268,"account_id":"aaa-bbb"}`
+	log2 := `{"time":"2026-03-11T12:00:00Z","level":"ERROR","msg":"search: no links found","file":"/app/tools/tool_web_search.go","line":268,"account_id":"ccc-ddd"}`
+	assert.Equal(t, NewPattern(log1).Hash(), NewPattern(log2).Hash(), "same msg with different metadata should produce same hash")
+
+	// Same "msg" with a large data blob must produce the same hash as without.
+	log3 := `{"time":"2026-03-10T05:02:40Z","level":"ERROR","msg":"search: no links found","data":"<!DOCTYPE html><html><head><meta charset='utf-8'><title>Brave Search</title></head><body><div class='results'>lots of html content here</div></body></html>"}`
+	assert.Equal(t, NewPattern(log1).Hash(), NewPattern(log3).Hash(), "data blob should not affect hash")
+
+	// Different "msg" must produce different hashes.
+	log4 := `{"time":"2026-03-10T05:02:40Z","level":"ERROR","msg":"rewoosolver: unable to unmarshal response"}`
+	assert.NotEqual(t, NewPattern(log1).Hash(), NewPattern(log4).Hash(), "different msg should produce different hash")
+
+	// None of these should produce the empty hash.
+	for _, l := range []string{log1, log2, log3, log4} {
+		assert.NotEqual(t, emptyHash, NewPattern(l).Hash(), "should not produce empty hash: %s", l[:80])
+	}
+}
+
+func TestJsonPatternCaseInsensitiveKeys(t *testing.T) {
+	// Keys should be matched case-insensitively.
+	log1 := `{"Message":"connection refused","Level":"ERROR"}`
+	log2 := `{"message":"connection refused","level":"error"}`
+	log3 := `{"MSG":"connection refused","LEVEL":"ERROR"}`
+	assert.Equal(t, NewPattern(log1).Hash(), NewPattern(log2).Hash())
+	assert.Equal(t, NewPattern(log1).Hash(), NewPattern(log3).Hash())
+}
+
+func TestJsonPatternFallback(t *testing.T) {
+	// JSON with no recognized message keys falls back to all values.
+	log := `{"component":"scheduler","action":"reconcile","status":"failed"}`
+	p := NewPattern(log)
+	assert.NotEqual(t, NewPattern("").Hash(), p.Hash(), "fallback should not produce empty hash")
+	assert.Contains(t, p.String(), "scheduler")
+}
+
+func TestJsonPatternWithErrorField(t *testing.T) {
+	// "error" field should be included in pattern extraction.
+	log1 := `{"msg":"request failed","error":"connection timeout"}`
+	log2 := `{"msg":"request failed","error":"permission denied"}`
+	assert.NotEqual(t, NewPattern(log1).Hash(), NewPattern(log2).Hash(),
+		"different error values should produce different hashes")
+}
+
+// TestJsonPatternRealLokiLogs validates the logparser against real production
+// JSON logs collected from Loki. These cover Go structured logs (zerolog),
+// Python structured logs, and logs with nested error objects/stacktraces.
+func TestJsonPatternRealLokiLogs(t *testing.T) {
+	emptyHash := NewPattern("").Hash()
+
+	// --- Go structured logs (zerolog "msg" key) ---
+
+	// Same msg, different metadata (account_id, trace_id, timestamps) → same hash.
+	goLog1 := `{"time":"2026-03-10T06:00:32.76936812Z","level":"ERROR","msg":"tool execution error","account_id":"a2a30b02-0f67-42e5-a2ab-c658230fd798","session_id":"094b32cb-f980-49a9-9152-c9c7c4b83349","user_id":"af4cb6af-1254-421d-bfa5-ffcfe649017e","tenant_id":"890cad87-c452-4aa7-b84a-742cee0454a1","trace_id":"d1540fb0a887608c10dbe8f6b37683f7","file":"/app/agents/core/executor_planner.go","line":2177,"error":"workspace /analyze call failed: proxy call failed: unknown","tool":"agent_code_2"}`
+	goLog2 := `{"time":"2026-03-10T07:15:00.000000000Z","level":"ERROR","msg":"tool execution error","account_id":"bbbbbbbb-0000-0000-0000-000000000000","session_id":"aaaaaaaa-0000-0000-0000-000000000000","user_id":"cccccccc-0000-0000-0000-000000000000","tenant_id":"dddddddd-0000-0000-0000-000000000000","trace_id":"eeeeeeee0000000000000000","file":"/app/agents/core/executor_planner.go","line":2177,"error":"workspace /analyze call failed: proxy call failed: unknown","tool":"agent_code_2"}`
+	assert.Equal(t, NewPattern(goLog1).Hash(), NewPattern(goLog2).Hash(),
+		"same msg+error with different metadata should produce same hash")
+	assert.NotEqual(t, emptyHash, NewPattern(goLog1).Hash())
+
+	// Different msg → different hash.
+	goLog3 := `{"time":"2026-03-10T06:00:32.769283951Z","level":"ERROR","msg":"code: failed to execute via workspace","account_id":"a2a30b02","trace_id":"d1540fb0","error":"workspace /analyze call failed: proxy call failed: unknown"}`
+	goLog4 := `{"time":"2026-03-10T06:00:32.769240731Z","level":"ERROR","msg":"workspace: proxy API call failed","account_id":"a2a30b02","trace_id":"d1540fb0","error":"unknown"}`
+	assert.NotEqual(t, NewPattern(goLog3).Hash(), NewPattern(goLog4).Hash(),
+		"different msg values should produce different hashes")
+
+	// Pattern should come from msg, not from metadata fields.
+	p1 := NewPattern(goLog1)
+	assert.Contains(t, p1.String(), "tool")
+	assert.Contains(t, p1.String(), "execution")
+	assert.Contains(t, p1.String(), "error")
+	assert.NotContains(t, p1.String(), "executor_planner.go",
+		"file paths should not leak into pattern")
+
+	// --- Go structured logs with nested error object (stringified as map) ---
+
+	goLogNested := `{"time":"2026-03-10T05:52:11.512955513Z","level":"ERROR","msg":"unable to fetch events","accountId":"fc9f8ff5-8c17-421e-b461-18e346832e6c","job_id":"e35f812b-9900-47e5-a31d-6f213becd52e","error":{"message":"failed to list audit events: request failed: context deadline exceeded","type":"*fmt.wrapError"}}`
+	pNested := NewPattern(goLogNested)
+	assert.NotEqual(t, emptyHash, pNested.Hash())
+	assert.Contains(t, pNested.String(), "unable")
+	assert.Contains(t, pNested.String(), "fetch")
+	assert.Contains(t, pNested.String(), "events")
+
+	// --- Go structured log: WARN with error field containing URL ---
+
+	goLogWarn := `{"time":"2026-03-10T06:05:44.826363378Z","level":"WARN","msg":"Primary model failed, analyzing error type","account_id":"a2a30b02","event_id":"d6dffec1","trace_id":"b1ccf7fe","error":"error in stream mode: Post \"https://generativelanguage.googleapis.com/v1beta/models/gemini-3-flash-preview:streamGenerateContent\": context deadline exceeded","model":"gemini-3-flash-preview"}`
+	pWarn := NewPattern(goLogWarn)
+	assert.NotEqual(t, emptyHash, pWarn.Hash())
+	assert.Contains(t, pWarn.String(), "Primary")
+	assert.Contains(t, pWarn.String(), "model")
+	assert.Contains(t, pWarn.String(), "failed")
+
+	// --- Python structured logs ("message" key + "exc_info" key) ---
+
+	pyLog := `{"asctime":"2026-03-10 06:00:44,782","levelname":"ERROR","filename":"search_logic.py","lineno":56,"message":"Error searching collection a2a30b02_long_term_memory: Unexpected Response: 400 (Bad Request)\nRaw response content:\nb'{\"status\":{\"error\":\"Wrong input: Vector dimension error: expected dim: 768, got 3072\"},\"time\":0.000263312}'","exc_info":"Traceback (most recent call last):\n  File \"/app/rag/search/search_logic.py\", line 36\n  response = client.query_points()\n"}`
+	pPy := NewPattern(pyLog)
+	assert.NotEqual(t, emptyHash, pPy.Hash())
+	assert.Contains(t, pPy.String(), "Error")
+	assert.Contains(t, pPy.String(), "searching")
+	assert.NotContains(t, pPy.String(), "search_logic.py",
+		"filename should not leak into pattern")
+	assert.NotContains(t, pPy.String(), "Traceback",
+		"exc_info should not leak into pattern")
+
+	// Same Python error from same collection → same hash even with different traceback.
+	pyLog2 := `{"asctime":"2026-03-10 07:00:00,000","levelname":"ERROR","filename":"other_file.py","lineno":99,"message":"Error searching collection a2a30b02_long_term_memory: Unexpected Response: 400 (Bad Request)\nRaw response content:\nb'{\"status\":{\"error\":\"Wrong input: Vector dimension error: expected dim: 768, got 3072\"},\"time\":0.000263312}'","exc_info":"different traceback here"}`
+	assert.Equal(t, pPy.Hash(), NewPattern(pyLog2).Hash(),
+		"same message with different traceback/filename should produce same hash")
+
+	// --- Python structured log: "message" key with Pydantic validation errors ---
+
+	pyLogPydantic := `{"asctime":"2026-03-10 06:00:11,298","levelname":"ERROR","filename":"message.py","lineno":870,"message":"Failed to send message: 6 validation errors for SLOAlertParams\nslo_target\n  Input should be a valid string [type=string_type, input_value=0.99, input_type=float]\nfiring_since\n  Input should be a valid string [type=string_type, input_value=1773122400.0, input_type=float]"}`
+	pPydantic := NewPattern(pyLogPydantic)
+	assert.NotEqual(t, emptyHash, pPydantic.Hash())
+	assert.Contains(t, pPydantic.String(), "Failed")
+	assert.Contains(t, pPydantic.String(), "send")
+	assert.Contains(t, pPydantic.String(), "message")
+	assert.NotContains(t, pPydantic.String(), "message.py",
+		"filename should not leak into pattern")
+
+	// --- JSON log with "message" key and large data blob ---
+
+	logWithDataBlob := `{"timestamp":"2026-03-10T06:07:07Z","log_type":"ERROR","event":"tool_failure","message":"Tool failed: cli","data":{"tool_name":"cli"},"duration":"5m39.826611881s","error":"would reformat /tmp/code-analysis--2474322462/nudgebee/notifications-server/notifications_server/message_templates/google_chat/auto_optimize_scheduled_notification.py\nwould reformat /tmp/code-analysis--2474322462/nudgebee/notifications-server/notifications_server/message_templates/google_chat/finding.py\nwould reformat many more files..."}`
+	logWithDataBlob2 := `{"timestamp":"2026-03-10T07:07:07Z","log_type":"ERROR","event":"tool_failure","message":"Tool failed: cli","data":{"tool_name":"cli"},"duration":"2m10.000000000s","error":"would reformat /tmp/code-analysis--9999999999/different/path/file.py"}`
+	pBlob1 := NewPattern(logWithDataBlob)
+	pBlob2 := NewPattern(logWithDataBlob2)
+	assert.NotEqual(t, emptyHash, pBlob1.Hash())
+	assert.NotEqual(t, emptyHash, pBlob2.Hash())
+	// Both have same "message" but different "error" — different hashes expected
+	// since error field is included in pattern extraction.
+	assert.Contains(t, pBlob1.String(), "Tool")
+	assert.Contains(t, pBlob1.String(), "failed")
+
+	// --- Stability: same msg from different pods/times → same hash ---
+
+	msgVariant1 := `{"time":"2026-03-10T05:52:56.777982952Z","level":"ERROR","msg":"pq: operator does not exist: uuid = text","tenant_id":"890cad87-c452-4aa7-b84a-742cee0454a1","user_id":"30b9833e-f667-4b0b-b2c1-065169968e24","trace_id":"d38476f460117ffcb7fece3edb16311a"}`
+	msgVariant2 := `{"time":"2026-03-11T12:00:00.000000000Z","level":"ERROR","msg":"pq: operator does not exist: uuid = text","tenant_id":"aaaaaaaa-0000-0000-0000-000000000000","user_id":"bbbbbbbb-0000-0000-0000-000000000000","trace_id":"cccccccc0000000000000000"}`
+	assert.Equal(t, NewPattern(msgVariant1).Hash(), NewPattern(msgVariant2).Hash(),
+		"identical msg from different pods/times must produce same hash")
+
+	// --- No empty hashes for any of these logs ---
+
+	allLogs := []string{goLog1, goLog2, goLog3, goLog4, goLogNested, goLogWarn,
+		pyLog, pyLog2, pyLogPydantic, logWithDataBlob, logWithDataBlob2,
+		msgVariant1, msgVariant2}
+	for i, l := range allLogs {
+		h := NewPattern(l).Hash()
+		assert.NotEqual(t, emptyHash, h, "log %d should not produce empty hash", i)
+		assert.NotEqual(t, "", NewPattern(l).String(), "log %d should not produce empty pattern", i)
+	}
+
+	// --- Verify metadata isolation: none of these patterns contain UUIDs or file paths ---
+
+	for i, l := range allLogs {
+		s := NewPattern(l).String()
+		assert.NotContains(t, s, "/app/", "log %d pattern should not contain file paths", i)
+		assert.NotContains(t, s, ".go", "log %d pattern should not contain .go files", i)
+		assert.NotContains(t, s, ".py", "log %d pattern should not contain .py files", i)
+	}
+}
+
+// TestJsonPatternBulkLokiLogs runs 459 unique production JSON logs through NewPattern
+// and asserts none produce empty hashes or panic. These were collected from Loki across
+// Go (zerolog), Python, and custom JSON log formats.
+func TestJsonPatternBulkLokiLogs(t *testing.T) {
+	// Read JSONL file: each line is a JSON-encoded string containing a raw log line.
+	data, err := os.ReadFile("/tmp/loki_test_logs.jsonl")
+	if err != nil {
+		t.Skip("bulk test logs not available at /tmp/loki_test_logs.jsonl — run Loki export first")
+	}
+
+	emptyHash := NewPattern("").Hash()
+	lines := bytes.Split(data, []byte("\n"))
+
+	var tested int
+	var emptyPatterns []int
+	hashCounts := map[string]int{}
+
+	for i, line := range lines {
+		if len(line) == 0 {
+			continue
+		}
+		// Each line is a JSON-encoded string — unwrap it.
+		var rawLog string
+		if err := json.Unmarshal(line, &rawLog); err != nil {
+			t.Logf("line %d: failed to unmarshal wrapper: %v", i, err)
+			continue
+		}
+
+		p := NewPattern(rawLog)
+		h := p.Hash()
+		s := p.String()
+		tested++
+
+		if h == emptyHash {
+			emptyPatterns = append(emptyPatterns, i)
+			t.Errorf("line %d produced empty hash (pattern=%q), log prefix: %s", i, s, rawLog[:min(len(rawLog), 120)])
+		}
+
+		hashCounts[h]++
+	}
+
+	t.Logf("Tested %d logs, %d unique hashes, %d empty-hash violations", tested, len(hashCounts), len(emptyPatterns))
+
+	// Report top hash collisions (most common patterns) for review.
+	type kv struct {
+		hash  string
+		count int
+	}
+	var top []kv
+	for h, c := range hashCounts {
+		if c > 1 {
+			top = append(top, kv{h, c})
+		}
+	}
+	sort.Slice(top, func(i, j int) bool { return top[i].count > top[j].count })
+	if len(top) > 10 {
+		top = top[:10]
+	}
+	for _, kv := range top {
+		t.Logf("  hash %s seen %d times", kv.hash, kv.count)
+	}
+}
+
+func min(a, b int) int {
+	if a < b {
+		return a
+	}
+	return b
 }