Add Unique() filter method

script.go

@@ -347,6 +347,28 @@ func (p *Pipe) Dirname() *Pipe {
 	})
 }
 
+// Unique returns a new pipe with consecutive duplicate lines suppressed.
+//
+// Unlike Freq(), which aggregates global counts across the entire stream,
+// Unique processes lines continuously in real time, keeping memory usage minimal
+// and preserving original chronological line order. Non-consecutive duplicates
+// (e.g. A, B, A) are retained.
+func (p *Pipe) Unique() *Pipe {
+	if p == nil || p.Error() != nil {
+		return p
+	}
+	var lastLine string
+	var seenAny bool
+
+	return p.FilterScan(func(line string, w io.Writer) {
+		if !seenAny || line != lastLine {
+			fmt.Fprintln(w, line)
+			seenAny = true
+			lastLine = line
+		}
+	})
+}
+
 // Do performs the HTTP request req using the pipe's configured HTTP client, as
 // set by [Pipe.WithHTTPClient], or [http.DefaultClient] otherwise. The
 // response body is streamed concurrently to the pipe's output. If the response

script_test.go

@@ -2738,3 +2738,55 @@ func ExampleSlice() {
 // methods that buffer input. We want to make sure they don't throw
 // "bufio.Scanner: token too long" errors.
 var longLine = strings.Repeat("super long line ", 4096) + "\nlast line\n"
+
+// TestUnique verifies that Pipe.Unique correctly collapses consecutive duplicate
+// lines while preserving line ordering and streaming behavior.
+//
+// Expected behavior:
+//   - Adjacent identical lines are reduced to a single line.
+//   - Non-adjacent duplicate lines (e.g. "a\nb\na\n") are preserved.
+//   - Unique streams and empty inputs pass through unmodified without errors.
+func TestUnique(t *testing.T) {
+	t.Parallel()
+
+	testCases := []struct {
+		name     string
+		input    string
+		expected string
+	}{
+		{
+			name:     "collapses consecutive duplicates while keeping non-consecutive ones",
+			input:    "a\na\nb\na\n",
+			expected: "a\nb\na\n",
+		},
+		{
+			name:     "collapses multiple consecutive occurrences",
+			input:    "apple\napple\napple\n",
+			expected: "apple\n",
+		},
+		{
+			name:     "passes through unique streams unchanged",
+			input:    "one\ntwo\nthree\n",
+			expected: "one\ntwo\nthree\n",
+		},
+		{
+			name:     "handles empty input safely",
+			input:    "",
+			expected: "",
+		},
+	}
+
+	for _, tc := range testCases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			got, err := script.Echo(tc.input).Unique().String()
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+			if got != tc.expected {
+				t.Errorf("Unique() on %q: got %q, want %q", tc.input, got, tc.expected)
+			}
+		})
+	}
+}