feat: discover NDJSON logs via message extraction

CONTEXT.md

@@ -27,6 +27,6 @@ _Avoid_: Sync, connection, integration, LogImport
 An external logging service LAPP can import from, such as GCP Cloud Logging or Vercel. LAPP uses credentials already present on the machine and never manages provider authentication itself.
 _Avoid_: Source, backend
 
-**Projection**:
-The text line derived from a structured log entry for pattern discovery. Discovery reads the projection; investigation material keeps the full structured entry.
-_Avoid_: Flattening, rendering
+**Extraction**:
+The text line pulled out of a structured log entry for pattern discovery. Discovery reads the extraction; investigation material keeps the full structured entry.
+_Avoid_: Projection, flattening, rendering

docs/adr/0006-ndjson-log-files-with-message-projection.md → docs/adr/0006-ndjson-log-files-with-message-extraction.md

@@ -1,16 +1,16 @@
-# Structured logs are stored as NDJSON and discovered via message projection
+# Structured logs are stored as NDJSON and discovered via message extraction
 
 Log files can be NDJSON (one JSON entry per line) as well as plain text, detected per file by the pipeline — the capability belongs to the pipeline, not to any import provider, so hand-uploaded NDJSON files get the same treatment as imported ones. Imported GCP entries land in a fixed envelope: `{"ts": ..., "severity": ..., "payload": {...}}` with the jsonPayload nested untouched (textPayload becomes `payload.message`); provider metadata such as labels, trace, and resource stays in the ImportRun record, not in the log lines.
 
-For pattern discovery, JSON entries are projected to a text line rather than fed to Drain whole: the first string field among `payload.message`, `msg`, `log`, `error` becomes the projection (`<severity> <message>`); if none exists, the compact payload JSON is the fallback. Timestamps are excluded from the projection to keep Drain templates clean.
+For pattern discovery, a text line is extracted from each JSON entry rather than feeding Drain the whole entry: the first string field among `payload.message`, `msg`, `log`, `error` becomes the extraction (`<severity> <message>`); if none exists, the compact payload JSON is the fallback. Timestamps are excluded from the extraction to keep Drain templates clean.
 
 **Considered Options**
 
 - Flatten everything to text at import time (structure lost)
 - NDJSON storage, discovery on compact-JSON lines (structure kept, dirty templates)
-- NDJSON storage with message projection (chosen)
+- NDJSON storage with message extraction (chosen)
 - JSON-native pattern discovery by structure/key-set (a second discovery engine — deferred, not rejected)
 
 **Consequences**
 
-The envelope and the projection rule are part of the workspace file contract (ADR 0001). Payload fields are never flattened to the top level, so user fields named `severity` or `ts` cannot collide with the envelope. Analysis agents can query structure with jq-style tools instead of grepping flattened text. If real usage shows dirty templates for payloads without a message-like field, the projection rule is the extension point.
+The envelope and the extraction rule are part of the workspace file contract (ADR 0001). Payload fields are never flattened to the top level, so user fields named `severity` or `ts` cannot collide with the envelope. Analysis agents can query structure with jq-style tools instead of grepping flattened text. If real usage shows dirty templates for payloads without a message-like field, the extraction rule is the extension point.

pkg/ndjson/detect.go

@@ -0,0 +1,48 @@
+// Package ndjson classifies log files as NDJSON and extracts the text lines
+// fed to pattern mining from structured entries, per ADR 0006.
+package ndjson
+
+import (
+	"encoding/json"
+	"strings"
+)
+
+// Format classifies the on-disk format of a log file.
+type Format string
+
+const (
+	// FormatText marks a file for the plain text pipeline.
+	FormatText Format = "text"
+	// FormatNDJSON marks a file whose lines are JSON objects, one per line.
+	FormatNDJSON Format = "ndjson"
+)
+
+// DetectFormat classifies a file's lines. A file is NDJSON only when it has
+// at least one non-empty line and every non-empty line parses as a JSON
+// object. Files mixing text with JSON lines stay on the text path, so the
+// extraction never has to guess on a half-structured file.
+func DetectFormat(lines []string) Format {
+	sampled := 0
+	for _, line := range lines {
+		trimmed := strings.TrimSpace(line)
+		if trimmed == "" {
+			continue
+		}
+		if !isJSONObject(trimmed) {
+			return FormatText
+		}
+		sampled++
+	}
+	if sampled == 0 {
+		return FormatText
+	}
+	return FormatNDJSON
+}
+
+func isJSONObject(trimmed string) bool {
+	if !strings.HasPrefix(trimmed, "{") {
+		return false
+	}
+	var obj map[string]any
+	return json.Unmarshal([]byte(trimmed), &obj) == nil
+}

pkg/ndjson/detect_test.go

@@ -0,0 +1,60 @@
+package ndjson
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+// TestDetectFormatFixtures classifies every fixture under testdata/detect/.
+// The expected format is encoded in the file name as <case>.expect-<format>.log,
+// so adding a case only means adding one fixture file.
+func TestDetectFormatFixtures(t *testing.T) {
+	paths, err := filepath.Glob(filepath.Join("testdata", "detect", "*.log"))
+	if err != nil {
+		t.Fatalf("glob detect fixtures: %v", err)
+	}
+	if len(paths) == 0 {
+		t.Fatal("no detect fixtures found under testdata/detect")
+	}
+
+	for _, path := range paths {
+		name, want := parseDetectFixtureName(t, path)
+		t.Run(name, func(t *testing.T) {
+			got := DetectFormat(readFixtureLines(t, path))
+			if got != want {
+				t.Fatalf("DetectFormat(%s) = %q, want %q", path, got, want)
+			}
+		})
+	}
+}
+
+func TestDetectFormatEmptyInput(t *testing.T) {
+	if got := DetectFormat(nil); got != FormatText {
+		t.Fatalf("DetectFormat(nil) = %q, want %q", got, FormatText)
+	}
+}
+
+func parseDetectFixtureName(t *testing.T, path string) (string, Format) {
+	t.Helper()
+	base := strings.TrimSuffix(filepath.Base(path), ".log")
+	name, expected, ok := strings.Cut(base, ".expect-")
+	if !ok {
+		t.Fatalf("detect fixture %s must be named <case>.expect-<format>.log", path)
+	}
+	format := Format(expected)
+	if format != FormatText && format != FormatNDJSON {
+		t.Fatalf("detect fixture %s has unknown expected format %q", path, expected)
+	}
+	return name, format
+}
+
+func readFixtureLines(t *testing.T, path string) []string {
+	t.Helper()
+	content, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("read fixture %s: %v", path, err)
+	}
+	return strings.Split(strings.TrimSuffix(string(content), "\n"), "\n")
+}

pkg/ndjson/extract.go

@@ -0,0 +1,80 @@
+package ndjson
+
+import (
+	"bytes"
+	"encoding/json"
+	"strings"
+)
+
+// messageFields are checked in order; the first string value wins.
+var messageFields = []string{"message", "msg", "log", "error"}
+
+// severityFields are checked in order; the first string value wins.
+// Numeric levels are ignored in v1.
+var severityFields = []string{"severity", "level"}
+
+// Extract converts one NDJSON line into the text line fed to pattern mining.
+// Envelope entries {"ts": ..., "severity": ..., "payload": {...}} extract
+// from the payload; any other object is the payload itself. The extracted line is
+// "<severity> <message>", just "<message>" when no severity-like string field
+// exists, or the compact payload JSON when no message-like string field
+// exists. A line that does not parse as a JSON object is returned unchanged.
+func Extract(line string) string {
+	var entry map[string]any
+	if err := json.Unmarshal([]byte(line), &entry); err != nil || entry == nil {
+		return line
+	}
+
+	payload, severity := splitEnvelope(entry)
+	if severity == "" {
+		severity = firstStringField(payload, severityFields)
+	}
+
+	message := firstStringField(payload, messageFields)
+	if message == "" {
+		return compactJSON(payload)
+	}
+	if severity == "" {
+		return message
+	}
+	return severity + " " + message
+}
+
+// splitEnvelope returns the payload object and the envelope severity when the
+// entry matches the fixed envelope shape, otherwise the entry itself. The
+// envelope is the importer's fixed contract (ADR 0006): ts, a string severity,
+// and a payload object must all be present; anything less is an arbitrary
+// user shape and is extracted as a whole.
+func splitEnvelope(entry map[string]any) (payload map[string]any, severity string) {
+	nested, ok := entry["payload"].(map[string]any)
+	if !ok {
+		return entry, ""
+	}
+	if _, hasTS := entry["ts"]; !hasTS {
+		return entry, ""
+	}
+	envelopeSeverity, ok := entry["severity"].(string)
+	if !ok || envelopeSeverity == "" {
+		return entry, ""
+	}
+	return nested, envelopeSeverity
+}
+
+func firstStringField(obj map[string]any, fields []string) string {
+	for _, field := range fields {
+		if value, ok := obj[field].(string); ok && value != "" {
+			return value
+		}
+	}
+	return ""
+}
+
+func compactJSON(payload map[string]any) string {
+	var buf bytes.Buffer
+	encoder := json.NewEncoder(&buf)
+	encoder.SetEscapeHTML(false)
+	if err := encoder.Encode(payload); err != nil {
+		return ""
+	}
+	return strings.TrimSuffix(buf.String(), "\n")
+}

pkg/ndjson/extract_test.go

@@ -0,0 +1,52 @@
+package ndjson
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+// TestExtractFixtures extracts every line of every input fixture under
+// testdata/extract/ and compares the result against its committed expected
+// output. Each <case>.input.ndjson pairs with <case>.expected.txt, so adding
+// a case only means adding one fixture pair.
+func TestExtractFixtures(t *testing.T) {
+	inputs, err := filepath.Glob(filepath.Join("testdata", "extract", "*.input.ndjson"))
+	if err != nil {
+		t.Fatalf("glob extraction fixtures: %v", err)
+	}
+	if len(inputs) == 0 {
+		t.Fatal("no extraction fixtures found under testdata/extract")
+	}
+
+	for _, inputPath := range inputs {
+		name := strings.TrimSuffix(filepath.Base(inputPath), ".input.ndjson")
+		expectedPath := filepath.Join("testdata", "extract", name+".expected.txt")
+		t.Run(name, func(t *testing.T) {
+			var extracted []string
+			for _, line := range readFixtureLines(t, inputPath) {
+				if strings.TrimSpace(line) == "" {
+					continue
+				}
+				extracted = append(extracted, Extract(line))
+			}
+			got := strings.Join(extracted, "\n") + "\n"
+
+			expected, err := os.ReadFile(expectedPath)
+			if err != nil {
+				t.Fatalf("read expected output %s: %v", expectedPath, err)
+			}
+			if got != string(expected) {
+				t.Fatalf("extraction mismatch for %s\ngot:\n%swant:\n%s", inputPath, got, expected)
+			}
+		})
+	}
+}
+
+func TestExtractNonJSONLineIsReturnedUnchanged(t *testing.T) {
+	line := "2026-06-06 10:00:00 INFO plain text line"
+	if got := Extract(line); got != line {
+		t.Fatalf("Extract(%q) = %q, want unchanged", line, got)
+	}
+}

pkg/ndjson/testdata/detect/json-arrays.expect-text.log

@@ -0,0 +1,2 @@
+[1,2,3]
+["a","b"]

pkg/ndjson/testdata/detect/mixed.expect-text.log

@@ -0,0 +1,4 @@
+{"ts":"2026-06-06T10:00:00Z","severity":"ERROR","payload":{"message":"db timeout user=42"}}
+{"level":"warn","msg":"queue depth 100"}
+2026-06-06 10:00:02 ERROR db timeout user=43
+{"event":"gc","duration_ms":12}

pkg/ndjson/testdata/detect/plain-text.expect-text.log

@@ -0,0 +1,3 @@
+2026-06-06 10:00:00 INFO server started port=8080
+2026-06-06 10:00:01 ERROR db timeout user=42
+2026-06-06 10:00:02 ERROR db timeout user=43

pkg/ndjson/testdata/detect/pure-ndjson.expect-ndjson.log

@@ -0,0 +1,5 @@
+{"ts":"2026-06-06T10:00:00Z","severity":"ERROR","payload":{"message":"db timeout user=42"}}
+{"level":"warn","msg":"queue depth 100"}
+
+{"event":"gc","duration_ms":12}
+{"ts":"2026-06-06T10:00:03Z","severity":"INFO","payload":{"message":"server started port=8080"}}

pkg/ndjson/testdata/extract/arbitrary-shape.expected.txt

@@ -0,0 +1,4 @@
+WARN disk usage high volume=/var
+connection reset peer=10.0.0.5
+context deadline exceeded
+primary text

pkg/ndjson/testdata/extract/arbitrary-shape.input.ndjson

@@ -0,0 +1,4 @@
+{"severity":"WARN","message":"disk usage high volume=/var"}
+{"log":"connection reset peer=10.0.0.5"}
+{"error":"context deadline exceeded","op":"fetch"}
+{"message":"primary text","msg":"secondary text"}

pkg/ndjson/testdata/extract/envelope.expected.txt

@@ -0,0 +1,3 @@
+ERROR db timeout user=42
+INFO server started port=8080
+{"payload":{"message":"heartbeat ok"},"ts":"2026-06-06T10:00:02Z"}

pkg/ndjson/testdata/extract/envelope.input.ndjson

@@ -0,0 +1,3 @@
+{"ts":"2026-06-06T10:00:00Z","severity":"ERROR","payload":{"message":"db timeout user=42"}}
+{"ts":"2026-06-06T10:00:01Z","severity":"INFO","payload":{"msg":"server started port=8080"}}
+{"ts":"2026-06-06T10:00:02Z","payload":{"message":"heartbeat ok"}}

pkg/ndjson/testdata/extract/no-message-fallback.expected.txt

@@ -0,0 +1,2 @@
+{"duration_ms":12,"event":"gc"}
+{"event":"cache_evict","keys":120}

pkg/ndjson/testdata/extract/no-message-fallback.input.ndjson

@@ -0,0 +1,2 @@
+{"event":"gc","duration_ms":12}
+{"ts":"2026-06-06T10:00:05Z","severity":"DEBUG","payload":{"event":"cache_evict","keys":120}}

pkg/ndjson/testdata/extract/numeric-level-ignored.expected.txt

@@ -0,0 +1,2 @@
+request done status=200
+upstream unavailable

pkg/ndjson/testdata/extract/numeric-level-ignored.input.ndjson

@@ -0,0 +1,2 @@
+{"level":30,"msg":"request done status=200"}
+{"level":50,"error":"upstream unavailable"}

pkg/ndjson/testdata/extract/severity-via-level.expected.txt

@@ -0,0 +1,2 @@
+warn queue depth 100
+info user login user=alice

pkg/ndjson/testdata/extract/severity-via-level.input.ndjson

@@ -0,0 +1,2 @@
+{"level":"warn","msg":"queue depth 100"}
+{"level":"info","message":"user login user=alice"}

pkg/workspace/builder.go

@@ -107,7 +107,7 @@ func (b *Builder) computePatterns() {
 	}
 	matches := make([]lineWithTemplate, 0, len(b.tagged))
 	for _, tl := range b.tagged {
-		t, ok := pattern.MatchTemplate(tl.Content, b.templates)
+		t, ok := pattern.MatchTemplate(tl.DrainLine(), b.templates)
 		id := ""
 		if ok {
 			id = t.ID.String()