fix(tail): fall back to raw line on malformed JSON instead of aborting

src/tail/eval_parse.zig

@@ -50,7 +50,10 @@ pub fn parseLogfmtAttrs(ctx: *context.TailLineContext, line: []const u8) !void {
 }
 
 pub fn parseJsonAttrs(ctx: *context.TailLineContext, line: []const u8) !void {
-    const parsed = try std.json.parseFromSliceLeaky(std.json.Value, ctx.allocator, line, .{});
+    const parsed = std.json.parseFromSliceLeaky(std.json.Value, ctx.allocator, line, .{}) catch |err| switch (err) {
+        error.OutOfMemory => return err,
+        else => return,
+    };
     if (parsed != .object) return;
 
     var it = parsed.object.iterator();
@@ -120,3 +123,22 @@ test "eval parse: logfmt extracts severity and attrs" {
     try testing.expectEqualStrings("INFO", ctx.severity.?);
     try testing.expectEqual(@as(usize, 3), ctx.attrs.items.len);
 }
+
+test "eval parse: malformed json falls back to raw line" {
+    var arena = std.heap.ArenaAllocator.init(testing.allocator);
+    defer arena.deinit();
+
+    const ctx = try parseLine(arena.allocator(), .json, "{not valid json");
+    try testing.expectEqualStrings("{not valid json", ctx.message.?);
+    try testing.expect(ctx.severity == null);
+    try testing.expectEqual(@as(usize, 0), ctx.attrs.items.len);
+}
+
+test "eval parse: truncated json falls back to raw line" {
+    var arena = std.heap.ArenaAllocator.init(testing.allocator);
+    defer arena.deinit();
+
+    const ctx = try parseLine(arena.allocator(), .json, "{\"message\":\"partia");
+    try testing.expectEqualStrings("{\"message\":\"partia", ctx.message.?);
+    try testing.expectEqual(@as(usize, 0), ctx.attrs.items.len);
+}

src/tail/eval_stream.zig

@@ -194,6 +194,37 @@ test "eval stream public API: json attribute matching and miss" {
     try testing.expect(try eval.evalLine("{\"message\":\"x\",\"ddsource\":\"app\"}"));
 }
 
+test "eval stream public API: malformed json line does not abort evaluation" {
+    var tmp = testing.tmpDir(.{});
+    defer tmp.cleanup();
+
+    const policy_path = try writePolicyFile(&tmp,
+        \\{
+        \\  "policies": [
+        \\    {
+        \\      "id": "drop-nginx",
+        \\      "name": "drop-nginx",
+        \\      "log": {
+        \\        "match": [{ "log_attribute": "ddsource", "regex": "^nginx$" }],
+        \\        "keep": "none"
+        \\      }
+        \\    }
+        \\  ]
+        \\}
+    );
+    defer testing.allocator.free(policy_path);
+
+    var stdio_bus = testBus();
+    var eval = try StreamEvaluator.init(testing.allocator, .json, policy_path, stdio_bus.eventBus());
+    defer eval.deinit();
+
+    try testing.expect(try eval.evalLine("{\"message\":\"x\",\"ddsource\":\"app\"}"));
+
+    try testing.expect(try eval.evalLine("{not valid json}"));
+
+    try testing.expect(!(try eval.evalLine("{\"message\":\"y\",\"ddsource\":\"nginx\"}")));
+}
+
 test "eval stream public API: logfmt severity matching" {
     var tmp = testing.tmpDir(.{});
     defer tmp.cleanup();