diff --git a/internal/search/matcher.go b/internal/search/matcher.go
index 6efd2b4..dbaf006 100644
--- a/internal/search/matcher.go
+++ b/internal/search/matcher.go
@@ -11,6 +11,10 @@ import (
 )
 
 // Matcher performs pattern matching on blob content according to search configurations.
+//
+// A Matcher is immutable after construction: nothing is written to it after
+// NewMatcher returns, and *regexp.Regexp is itself safe for concurrent use, so
+// one Matcher may be shared by every pipeline worker.
 type Matcher struct {
 	re            *regexp.Regexp
 	cfg           *model.Config
diff --git a/internal/search/pipeline.go b/internal/search/pipeline.go
index 00ba063..8c7b849 100644
--- a/internal/search/pipeline.go
+++ b/internal/search/pipeline.go
@@ -53,6 +53,10 @@ func isBlobReadError(err error) bool {
 }
 
 // Pipeline coordinates concurrent blob decompression, search matching, and provenance association.
+//
+// A Pipeline is immutable after construction: its reader, matcher, config, and
+// worker count are never written after NewPipeline returns, so a single
+// Pipeline is safe for concurrent use by multiple goroutines.
 type Pipeline struct {
 	reader  gitengine.ObjectReader
 	matcher *Matcher
@@ -80,8 +84,18 @@ type blobTask struct {
 }
 
 // ExecuteContext executes the search pipeline with context cancellation support,
-// bounded worker pools, channel backpressure, and zero goroutine leaks.
-// It streams results and errors over bounded channels until completion or cancellation.
+// bounded worker pools, and channel backpressure. It streams results and errors
+// over bounded channels until completion or cancellation.
+//
+// The pipeline leaks no goroutines provided the caller either drains resultsCh to
+// completion or cancels ctx: workers block on resultsCh sends and unblock only on
+// a receive or on cancellation. resultsCh is closed once every worker has exited,
+// and errCh is closed immediately after, so a caller may read errCh once the
+// range over resultsCh ends.
+//
+// Work runs under a cancellable child of ctx, so the pipeline stops early on the
+// first fatal error and, in quiet mode, on the first match. Neither is a
+// cancellation error: only cancellation of ctx itself is reported on errCh.
 func (p *Pipeline) ExecuteContext(ctx context.Context, occurrences []model.BlobOccurrence) (<-chan *BlobResult, <-chan error) {
 	if ctx == nil {
 		ctx = context.Background()
@@ -130,70 +144,52 @@ func (p *Pipeline) ExecuteContext(ctx context.Context, occurrences []model.BlobO
 
 	tasksCh := make(chan *blobTask, bufSize)
 	resultsCh := make(chan *BlobResult, bufSize)
+	// errCh has capacity 1 and every send on it is non-blocking, which implements
+	// first-fatal-error-wins: the first fatal error is kept and later ones are
+	// dropped. That capacity is exactly what guarantees a worker never blocks
+	// publishing an error, because the consumer usually does not read errCh until
+	// resultsCh has been drained. Raising it "so no error is lost" would let a
+	// worker's error send outlive the consumer's interest and reintroduce that
+	// deadlock.
 	errCh := make(chan error, 1)
 
-	// Dispatcher feeding tasks into bounded tasksCh with ctx cancellation check
-	go func() {
-		defer close(tasksCh)
-		for _, task := range jobOrder {
-			select {
-			case <-ctx.Done():
-				return
-			case tasksCh <- task:
-			}
-		}
-	}()
+	// All work runs under a cancellable child of the caller's context so that the
+	// first fatal error and a quiet-mode match can abandon the remaining blobs at
+	// once. callerCtx is kept for the closer: an internal early stop must never be
+	// reported as a cancellation.
+	callerCtx := ctx
+	ctx, cancelWork := context.WithCancel(callerCtx)
+
+	go p.dispatch(ctx, jobOrder, tasksCh)
 
-	// Launch worker pool
-	var stop atomic.Bool
+	// searched counts tasks whose result was fully delivered. It tells the closer
+	// whether the run actually finished, so a cancellation arriving after the last
+	// result was handed over cannot turn a complete result set into a failure.
+	var searched atomic.Int64
 	var wg sync.WaitGroup
-	for i := 0; i < p.workers; i++ {
+	for range p.workers {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			for {
-				select {
-				case <-ctx.Done():
-					return
-				case task, ok := <-tasksCh:
-					if !ok {
-						return
-					}
-					if stop.Load() {
-						continue
-					}
-					res := p.processTask(ctx, task)
-					// Soft per-blob read failures are delivered on resultsCh only; every
-					// other error is fatal and also reported on errCh.
-					if res.Error != nil && !isBlobReadError(res.Error) {
-						select {
-						case errCh <- res.Error:
-						default:
-						}
-					}
-					if p.cfg.Quiet && (len(res.Matches) > 0 || res.IsBinary) {
-						stop.Store(true)
-					}
-					// Only send results that have matches, are binary, or have errors
-					if len(res.Matches) > 0 || res.IsBinary || res.Error != nil {
-						select {
-						case <-ctx.Done():
-							return
-						case resultsCh <- res:
-						}
-					}
-				}
-			}
+			p.runWorker(ctx, cancelWork, &searched, tasksCh, resultsCh, errCh)
 		}()
 	}
 
-	// Closer goroutine waits for all workers to exit, captures cancellation error, and closes channels
+	// The closer is the sole owner of resultsCh and errCh: it waits for every
+	// worker to exit, publishes a caller-side cancellation, and closes both. The
+	// publication is gated on callerCtx, never on the derived ctx, because a quiet
+	// early stop or a fatal error cancels the derived one and must not surface as
+	// context.Canceled. It is gated on searched as well, because a caller who
+	// received every result was not cut short, whenever the cancel arrived.
 	go func() {
+		defer cancelWork()
 		wg.Wait()
-		if err := ctx.Err(); err != nil {
-			select {
-			case errCh <- err:
-			default:
+		if searched.Load() < int64(len(jobOrder)) {
+			if err := callerCtx.Err(); err != nil {
+				select {
+				case errCh <- err:
+				default:
+				}
 			}
 		}
 		close(resultsCh)
@@ -203,6 +199,78 @@ func (p *Pipeline) ExecuteContext(ctx context.Context, occurrences []model.BlobO
 	return resultsCh, errCh
 }
 
+// dispatch feeds every deduplicated task into tasksCh and is its only sender and
+// closer. It abandons the remaining tasks as soon as ctx is cancelled, which is
+// how a quiet-mode match, a fatal error, or a caller cancellation stops the run
+// without draining jobOrder.
+func (p *Pipeline) dispatch(ctx context.Context, jobOrder []*blobTask, tasksCh chan<- *blobTask) {
+	defer close(tasksCh)
+	for _, task := range jobOrder {
+		select {
+		case <-ctx.Done():
+			return
+		case tasksCh <- task:
+		}
+	}
+}
+
+// runWorker searches tasks until tasksCh is drained and closed or ctx is
+// cancelled. It never closes any channel it is given; cancelWork cancels ctx to
+// stop the whole run after the first fatal error and, in quiet mode, after the
+// first match. Every task whose result reaches resultsCh, or that has no result
+// to report, is counted in searched; a task abandoned to cancellation is not.
+func (p *Pipeline) runWorker(
+	ctx context.Context,
+	cancelWork context.CancelFunc,
+	searched *atomic.Int64,
+	tasksCh <-chan *blobTask,
+	resultsCh chan<- *BlobResult,
+	errCh chan<- error,
+) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case task, ok := <-tasksCh:
+			if !ok {
+				return
+			}
+			res := p.safeProcessTask(ctx, task)
+			if ctx.Err() != nil {
+				// The run is already being torn down, by the caller or by another
+				// worker. res is redundant, and any error it carries is just that
+				// cancellation, which the closer reports if the caller caused it.
+				return
+			}
+			// Soft per-blob read failures are delivered on resultsCh only; every
+			// other error is fatal: it is reported on errCh and stops the run.
+			fatal := res.Error != nil && !isBlobReadError(res.Error)
+			if fatal {
+				select {
+				case errCh <- res.Error:
+				default:
+				}
+			}
+			// Only send results that have matches, are binary, or have errors
+			if len(res.Matches) > 0 || res.IsBinary || res.Error != nil {
+				select {
+				case <-ctx.Done():
+					return
+				case resultsCh <- res:
+				}
+			}
+			searched.Add(1)
+			// The result is published, so the remaining blobs are now pointless:
+			// -q wants nothing beyond the first hit, and a fatal error abandons the
+			// run. Cancelling unblocks the dispatcher and the other workers.
+			if fatal || (p.cfg.Quiet && (len(res.Matches) > 0 || res.IsBinary)) {
+				cancelWork()
+				return
+			}
+		}
+	}
+}
+
 // Execute provides backwards-compatible synchronous execution by executing with context.Background()
 // and collecting all results into a slice in the original deduplicated order.
 func (p *Pipeline) Execute(occurrences []model.BlobOccurrence) ([]*BlobResult, error) {
@@ -229,9 +297,24 @@ func (p *Pipeline) Execute(occurrences []model.BlobOccurrence) ([]*BlobResult, e
 	return results, nil
 }
 
-// ExecuteStream streams search results using context.Background().
-func (p *Pipeline) ExecuteStream(occurrences []model.BlobOccurrence) (<-chan *BlobResult, <-chan error) {
-	return p.ExecuteContext(context.Background(), occurrences)
+// safeProcessTask turns a panic in processTask into an ordinary fatal error.
+// processTask drives packfile index lookups, zlib inflate, and delta
+// reconstruction over bytes from an arbitrary .git directory; a panic on a worker
+// goroutine cannot be recovered by the caller and would kill the process, taking
+// the reader's cleanup and the exit-code contract with it. The recover is scoped
+// to a single task so one poisoned blob does not stop the worker from reporting
+// it through the normal fatal-error path.
+func (p *Pipeline) safeProcessTask(ctx context.Context, task *blobTask) (res *BlobResult) {
+	defer func() {
+		if r := recover(); r != nil {
+			res = &BlobResult{
+				BlobOID:     task.oid,
+				Occurrences: task.occurrences,
+				Error:       fmt.Errorf("panic searching blob %s: %v", task.oid, r),
+			}
+		}
+	}()
+	return p.processTask(ctx, task)
 }
 
 // processTask reads the blob from Git object store and applies binary detection and pattern matching.
diff --git a/internal/search/pipeline_test.go b/internal/search/pipeline_test.go
index bc1fb27..04c3cbc 100644
--- a/internal/search/pipeline_test.go
+++ b/internal/search/pipeline_test.go
@@ -7,6 +7,8 @@ import (
 	"errors"
 	"fmt"
 	"runtime"
+	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -465,3 +467,240 @@ func TestPipelineSkipsUnreadableBlob(t *testing.T) {
 		})
 	}
 }
+
+// countingReader records how many blobs the pipeline actually read, and panics
+// on selected OIDs to stand in for a corrupt object that trips a bug in inflate
+// or delta reconstruction.
+type countingReader struct {
+	*mockSearchReader
+	reads   atomic.Int64
+	panicOn map[string]bool // filled in before ExecuteContext, never written after
+}
+
+func newCountingReader() *countingReader {
+	return &countingReader{
+		mockSearchReader: newMockSearchReader(),
+		panicOn:          make(map[string]bool),
+	}
+}
+
+func (c *countingReader) ReadObject(oid string) (*gitengine.Object, error) {
+	c.reads.Add(1)
+	if c.panicOn[oid] {
+		panic("simulated corrupt object " + oid)
+	}
+	return c.mockSearchReader.ReadObject(oid)
+}
+
+// matchingOccurrences stores n distinct blobs that all match the pattern
+// "needle" and returns one occurrence for each, in blob order.
+func matchingOccurrences(r *countingReader, n int) []model.BlobOccurrence {
+	occurrences := make([]model.BlobOccurrence, 0, n)
+	for i := range n {
+		oid := r.putBlob([]byte(fmt.Sprintf("needle in blob %d\n", i)))
+		occurrences = append(occurrences, model.BlobOccurrence{
+			BlobOID:   oid,
+			Path:      fmt.Sprintf("file_%d.txt", i),
+			CommitSHA: "c1",
+			Mode:      0100644,
+		})
+	}
+	return occurrences
+}
+
+// -q asks for the first hit only. The pipeline must abandon the remaining blobs
+// instead of pulling all of them through the queue, and its early stop must not
+// be reported as a cancellation error.
+func TestPipelineQuietStopsAfterFirstMatch(t *testing.T) {
+	t.Run("single worker reads only the first blob", func(t *testing.T) {
+		reader := newCountingReader()
+		occurrences := matchingOccurrences(reader, 200)
+
+		cfg := &model.Config{Pattern: "needle", Quiet: true}
+		matcher, err := NewMatcher(cfg)
+		if err != nil {
+			t.Fatalf("NewMatcher failed: %v", err)
+		}
+		p := NewPipeline(reader, matcher, cfg)
+		p.workers = 1 // one worker, so the first task alone decides the run
+
+		resultsCh, errCh := p.ExecuteContext(context.Background(), occurrences)
+		var results []*BlobResult
+		for res := range resultsCh {
+			results = append(results, res)
+		}
+		if err := <-errCh; err != nil {
+			t.Fatalf("quiet early stop must not be an error, got %v", err)
+		}
+		if len(results) != 1 {
+			t.Errorf("got %d results, want 1: -q reports a single hit", len(results))
+		}
+		if got := reader.reads.Load(); got != 1 {
+			t.Errorf("read %d blobs, want exactly 1: -q must stop at the first match", got)
+		}
+	})
+
+	t.Run("concurrent workers stop well before the queue is drained", func(t *testing.T) {
+		reader := newCountingReader()
+		occurrences := matchingOccurrences(reader, 2000)
+
+		cfg := &model.Config{Pattern: "needle", Quiet: true}
+		matcher, err := NewMatcher(cfg)
+		if err != nil {
+			t.Fatalf("NewMatcher failed: %v", err)
+		}
+
+		resultsCh, errCh := NewPipeline(reader, matcher, cfg).ExecuteContext(context.Background(), occurrences)
+		results := 0
+		for range resultsCh {
+			results++
+		}
+		if err := <-errCh; err != nil {
+			t.Fatalf("quiet early stop must not be an error, got %v", err)
+		}
+		if results == 0 {
+			t.Errorf("expected at least the one hit -q asks for")
+		}
+		// Cancellation bounds the reads to the tasks already buffered or in
+		// flight; without it every one of the 2000 blobs is pulled from the queue.
+		if got, limit := reader.reads.Load(), int64(len(occurrences)/2); got > limit {
+			t.Errorf("read %d of %d blobs, want at most %d: -q must cancel, not drain", got, len(occurrences), limit)
+		}
+	})
+}
+
+// A fatal error abandons the run. The pipeline must stop at once instead of
+// searching every remaining blob and only revealing the failure when resultsCh
+// closes, and the failure it reports must be the error itself, not the
+// cancellation that error triggered.
+func TestPipelineFatalErrorStopsRunEarly(t *testing.T) {
+	reader := newCountingReader()
+	occurrences := matchingOccurrences(reader, 300)
+	reader.panicOn[occurrences[0].BlobOID] = true
+
+	cfg := &model.Config{Pattern: "needle"}
+	matcher, err := NewMatcher(cfg)
+	if err != nil {
+		t.Fatalf("NewMatcher failed: %v", err)
+	}
+	p := NewPipeline(reader, matcher, cfg)
+	p.workers = 1 // one worker, so the failing blob is the first task
+
+	resultsCh, errCh := p.ExecuteContext(context.Background(), occurrences)
+	for range resultsCh {
+	}
+
+	err = <-errCh
+	if err == nil {
+		t.Fatalf("a fatal error must be reported on errCh")
+	}
+	if errors.Is(err, context.Canceled) {
+		t.Errorf("stopping the run must not mask the fatal error as a cancellation: %v", err)
+	}
+	if got := reader.reads.Load(); got != 1 {
+		t.Errorf("read %d of %d blobs, want exactly 1: a fatal error must stop the run", got, len(occurrences))
+	}
+}
+
+// The pipeline inflates and reconstructs bytes from an arbitrary .git
+// directory, so a panic must become an ordinary fatal error naming the offending
+// blob instead of killing the process (here, the test binary) and skipping every
+// deferred cleanup and the exit-code contract.
+func TestPipelinePanicBecomesFatalError(t *testing.T) {
+	reader := newCountingReader()
+	good := reader.putBlob([]byte("needle in a readable blob\n"))
+	poison := reader.putBlob([]byte("needle in a poisoned blob\n"))
+	reader.panicOn[poison] = true
+
+	occurrences := []model.BlobOccurrence{
+		{BlobOID: good, Path: "good.txt", CommitSHA: "c1", Mode: 0100644},
+		{BlobOID: poison, Path: "poison.txt", CommitSHA: "c1", Mode: 0100644},
+	}
+
+	cfg := &model.Config{Pattern: "needle"}
+	matcher, err := NewMatcher(cfg)
+	if err != nil {
+		t.Fatalf("NewMatcher failed: %v", err)
+	}
+	p := NewPipeline(reader, matcher, cfg)
+	p.workers = 1 // one worker, so the readable blob is searched before the panic
+
+	resultsCh, errCh := p.ExecuteContext(context.Background(), occurrences)
+	byOID := make(map[string]*BlobResult, len(occurrences))
+	for res := range resultsCh {
+		byOID[res.BlobOID] = res
+	}
+
+	err = <-errCh
+	if err == nil {
+		t.Fatalf("a panicking blob must surface as a fatal error on errCh")
+	}
+	if !strings.Contains(err.Error(), poison) {
+		t.Errorf("fatal error must name the offending blob %s, got: %v", poison, err)
+	}
+
+	res, ok := byOID[poison]
+	if !ok {
+		t.Fatalf("expected a result carrying the panic for %s", poison)
+	}
+	if res.Error == nil || !strings.Contains(res.Error.Error(), "panic") {
+		t.Errorf("result for the poisoned blob must carry the recovered panic, got %v", res.Error)
+	}
+	if isBlobReadError(res.Error) {
+		t.Errorf("a panic is fatal, not a soft per-blob read failure: %v", res.Error)
+	}
+
+	res, ok = byOID[good]
+	if !ok {
+		t.Fatalf("the blob searched before the panic must still be delivered")
+	}
+	if len(res.Matches) != 1 {
+		t.Errorf("got %d matches for the readable blob, want 1", len(res.Matches))
+	}
+}
+
+// A cancellation arriving once every result has been handed to the caller must
+// not turn a complete run into a failure: nothing was cut short, so exit codes
+// derived from errCh must stay clean even if the signal lands microseconds after
+// the last worker finished.
+func TestPipelineCancelAfterCompletion(t *testing.T) {
+	reader := newCountingReader()
+	occurrences := matchingOccurrences(reader, 64)
+
+	cfg := &model.Config{Pattern: "needle"}
+	matcher, err := NewMatcher(cfg)
+	if err != nil {
+		t.Fatalf("NewMatcher failed: %v", err)
+	}
+	p := NewPipeline(reader, matcher, cfg)
+
+	// Repeat so the cancel lands at varying points of the shutdown sequence.
+	for range 30 {
+		func() {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			resultsCh, errCh := p.ExecuteContext(ctx, occurrences)
+
+			// Take every expected result, then cancel while the workers and the
+			// closer may still be winding down.
+			for got := 0; got < len(occurrences); got++ {
+				res, ok := <-resultsCh
+				if !ok {
+					t.Fatalf("resultsCh closed after %d of %d results", got, len(occurrences))
+				}
+				if res.Error != nil {
+					t.Fatalf("unexpected result error: %v", res.Error)
+				}
+			}
+			cancel()
+
+			for range resultsCh {
+				t.Errorf("got more results than the %d blobs searched", len(occurrences))
+			}
+			if err := <-errCh; err != nil {
+				t.Fatalf("cancel after the full result set must leave errCh empty, got %v", err)
+			}
+		}()
+	}
+}