diff --git a/cmd/pipelines/test_events.go b/cmd/pipelines/test_events.go
new file mode 100644
index 00000000000..e9dd88aff70
--- /dev/null
+++ b/cmd/pipelines/test_events.go
@@ -0,0 +1,317 @@
+package pipelines
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/databricks/cli/libs/auth"
+	databricks "github.com/databricks/databricks-sdk-go"
+	"github.com/databricks/databricks-sdk-go/client"
+	"github.com/databricks/databricks-sdk-go/service/pipelines"
+)
+
+const (
+	testCaseProgressEventType = "test_case_progress"
+	testRunSummaryEventType   = "test_run_summary"
+)
+
+type testCaseStatus string
+
+const (
+	testCasePending testCaseStatus = "PENDING"
+	testCaseRunning testCaseStatus = "RUNNING"
+	testCasePassed  testCaseStatus = "PASSED"
+	testCaseFailed  testCaseStatus = "FAILED"
+	testCaseSkipped testCaseStatus = "SKIPPED"
+	testCaseError   testCaseStatus = "ERROR"
+)
+
+// parseTestCaseStatus accepts both the short status and the proto enum value name.
+func parseTestCaseStatus(s string) testCaseStatus {
+	return testCaseStatus(strings.TrimPrefix(s, "PIPELINE_TEST_CASE_STATUS_"))
+}
+
+func (s testCaseStatus) isFinal() bool {
+	switch s {
+	case testCasePassed, testCaseFailed, testCaseSkipped, testCaseError:
+		return true
+	default:
+		return false
+	}
+}
+
+type testRunSummary struct {
+	Total   int `json:"total"`
+	Passed  int `json:"passed"`
+	Failed  int `json:"failed"`
+	Skipped int `json:"skipped"`
+	Errored int `json:"errored"`
+}
+
+// testPipelineEvent is an entry of List Pipeline Events. The Go SDK PipelineEvent has no
+// details or origin.test_case_id, so test events are decoded from the raw response.
+type testPipelineEvent struct {
+	Id        string `json:"id"`
+	EventType string `json:"event_type"`
+	Timestamp string `json:"timestamp"`
+	Origin    struct {
+		TestCaseId string `json:"test_case_id"`
+	} `json:"origin"`
+	Error      *pipelines.ErrorDetail `json:"error"`
+	Truncation *pipelines.Truncation  `json:"truncation"`
+	Details    struct {
+		TestCaseProgress *struct {
+			Status string `json:"status"`
+		} `json:"test_case_progress"`
+		TestRunSummary *testRunSummary `json:"test_run_summary"`
+	} `json:"details"`
+}
+
+func (e testPipelineEvent) isTruncated(field string) bool {
+	if e.Truncation == nil {
+		return false
+	}
+	for _, f := range e.Truncation.TruncatedFields {
+		if f.FieldName == field {
+			return true
+		}
+	}
+	return false
+}
+
+// testEventsPageSize is the largest page List Pipeline Events returns.
+const testEventsPageSize = 250
+
+// readTestEvents returns the events of one update in timestamp order, following every page. With
+// sinceTimestamp it returns only the events at or after that timestamp, which must be a timestamp
+// string exactly as the API returned it.
+func readTestEvents(ctx context.Context, w *databricks.WorkspaceClient, pipelineID, updateID, sinceTimestamp string) ([]testPipelineEvent, error) {
+	apiClient, err := client.New(w.Config)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create API client: %w", err)
+	}
+	path := fmt.Sprintf("/api/2.0/pipelines/%s/events", pipelineID)
+	request := pipelines.ListPipelineEventsRequest{
+		Filter:     buildPipelineEventFilter(updateID, nil, nil, sinceTimestamp, ""),
+		OrderBy:    []string{"timestamp asc"},
+		MaxResults: testEventsPageSize,
+	}
+	var events []testPipelineEvent
+	for {
+		var page struct {
+			Events        []testPipelineEvent `json:"events"`
+			NextPageToken string              `json:"next_page_token"`
+		}
+		err := apiClient.Do(ctx, http.MethodGet, path, auth.WorkspaceIDHeaders(w.Config), nil, request, &page)
+		if err != nil {
+			return nil, fmt.Errorf("failed to fetch pipeline events: %w", err)
+		}
+		events = append(events, page.Events...)
+		if page.NextPageToken == "" || len(page.Events) == 0 {
+			return events, nil
+		}
+		// A request with page_token must not repeat filter or order_by: the token carries them,
+		// and the service rejects the request otherwise.
+		request = pipelines.ListPipelineEventsRequest{PageToken: page.NextPageToken, MaxResults: testEventsPageSize}
+	}
+}
+
+// testCaseResult is what the events report about one test case.
+type testCaseResult struct {
+	// id is origin.test_case_id: the absolute workspace path of the test file followed by the pytest
+	// test ID, such as /Workspace/Users/someone@example.com/orders/tests/test_orders.py::test_backfill.
+	id     string
+	status testCaseStatus
+	// finals counts final outcomes; a complete result has exactly one per case.
+	finals           int
+	started          time.Time
+	finished         time.Time
+	errorDetail      *pipelines.ErrorDetail
+	detailsTruncated bool
+}
+
+// duration returns the time between the RUNNING and the final event, if both are known.
+func (c *testCaseResult) duration() (time.Duration, bool) {
+	if c.started.IsZero() || c.finished.IsZero() || c.finished.Before(c.started) {
+		return 0, false
+	}
+	return c.finished.Sub(c.started), true
+}
+
+// testRun accumulates the events of a test update, ignoring events it has already seen.
+type testRun struct {
+	seen      map[string]bool
+	cases     []*testCaseResult
+	byID      map[string]*testCaseResult
+	summaries []testRunSummary
+	// unreadable counts test events whose outcome is missing.
+	unreadable int
+	// fatalError is the first fatal error that the update reported, other than the test verdict.
+	fatalError string
+	// verdictError is the service's verdict that tests failed, if the update reported it.
+	verdictError string
+}
+
+// testVerdictErrorPrefix starts the message of the error by which the service reports that tests
+// failed, such as "[PIPELINE_TEST_FAILED] 1 of the pipeline's tests failed (1 passed, 0 skipped).".
+// The message is matched because ErrorDetail has no error class yet; only the output depends on it.
+const testVerdictErrorPrefix = "[PIPELINE_TEST_FAILED]"
+
+// isTestVerdict reports whether an update error is the service's verdict that tests failed.
+func isTestVerdict(e *pipelines.ErrorDetail) bool {
+	for _, exception := range e.Exceptions {
+		if strings.HasPrefix(exception.Message, testVerdictErrorPrefix) {
+			return true
+		}
+	}
+	return false
+}
+
+func newTestRun() *testRun {
+	return &testRun{seen: map[string]bool{}, byID: map[string]*testCaseResult{}}
+}
+
+// add records an event and returns the case it gave a final outcome to, if any.
+func (r *testRun) add(e testPipelineEvent) *testCaseResult {
+	if r.seen[e.Id] {
+		return nil
+	}
+	r.seen[e.Id] = true
+
+	if e.Error != nil && e.Error.Fatal {
+		switch {
+		case isTestVerdict(e.Error):
+			r.verdictError = errorDetailText(e.Error)
+		case r.fatalError == "":
+			r.fatalError = errorDetailText(e.Error)
+		}
+	}
+
+	switch e.EventType {
+	case testRunSummaryEventType:
+		if e.Details.TestRunSummary == nil {
+			r.unreadable++
+			return nil
+		}
+		r.summaries = append(r.summaries, *e.Details.TestRunSummary)
+	case testCaseProgressEventType:
+		progress := e.Details.TestCaseProgress
+		if progress == nil || e.Origin.TestCaseId == "" {
+			r.unreadable++
+			return nil
+		}
+		status := parseTestCaseStatus(progress.Status)
+		if !status.isFinal() && status != testCasePending && status != testCaseRunning {
+			r.unreadable++
+			return nil
+		}
+
+		c, ok := r.byID[e.Origin.TestCaseId]
+		if !ok {
+			c = &testCaseResult{id: e.Origin.TestCaseId, status: testCasePending}
+			r.byID[c.id] = c
+			r.cases = append(r.cases, c)
+		}
+		timestamp, _ := time.Parse(time.RFC3339Nano, e.Timestamp)
+		switch {
+		case status == testCaseRunning:
+			c.started = timestamp
+			if !c.status.isFinal() {
+				c.status = testCaseRunning
+			}
+		case status.isFinal():
+			c.finals++
+			c.status = status
+			c.finished = timestamp
+			c.errorDetail = e.Error
+			c.detailsTruncated = e.isTruncated("error")
+			if c.finals == 1 {
+				return c
+			}
+		}
+	}
+	return nil
+}
+
+// counts returns the number of cases with each final outcome.
+func (r *testRun) counts() testRunSummary {
+	var s testRunSummary
+	for _, c := range r.cases {
+		if !c.status.isFinal() {
+			continue
+		}
+		s.Total++
+		switch c.status {
+		case testCasePassed:
+			s.Passed++
+		case testCaseFailed:
+			s.Failed++
+		case testCaseSkipped:
+			s.Skipped++
+		case testCaseError:
+			s.Errored++
+		default:
+		}
+	}
+	return s
+}
+
+// incompleteReason explains why the events of an ended update are not a complete result,
+// or returns "" if they are: one summary and one final outcome per case, matching its counts.
+func (r *testRun) incompleteReason(state pipelines.UpdateInfoState) string {
+	if r.unreadable > 0 {
+		return "some test results could not be read"
+	}
+	switch len(r.summaries) {
+	case 0:
+		switch state {
+		case pipelines.UpdateInfoStateFailed:
+			return "the update failed before all test results were received"
+		case pipelines.UpdateInfoStateCanceled:
+			return "the update was canceled before all test results were received"
+		default:
+			return "the update ended without a test summary"
+		}
+	case 1:
+	default:
+		return fmt.Sprintf("received %d test summaries instead of one", len(r.summaries))
+	}
+
+	for _, c := range r.cases {
+		if c.finals > 1 {
+			return fmt.Sprintf("test %s has more than one result", c.id)
+		}
+	}
+	summary := r.summaries[0]
+	received := r.counts()
+	if received.Total != summary.Total {
+		return fmt.Sprintf("received %d of %d test results", received.Total, summary.Total)
+	}
+	if received != summary {
+		return "the test results do not match the test summary"
+	}
+	for _, c := range r.cases {
+		if c.finals == 0 {
+			return fmt.Sprintf("test %s has no result", c.id)
+		}
+	}
+	return ""
+}
+
+func errorDetailText(detail *pipelines.ErrorDetail) string {
+	var parts []string
+	for _, e := range detail.Exceptions {
+		switch {
+		case e.ClassName != "" && e.Message != "":
+			parts = append(parts, e.ClassName+": "+e.Message)
+		case e.Message != "":
+			parts = append(parts, e.Message)
+		case e.ClassName != "":
+			parts = append(parts, e.ClassName)
+		}
+	}
+	return strings.Join(parts, "\n")
+}
diff --git a/cmd/pipelines/test_events_test.go b/cmd/pipelines/test_events_test.go
new file mode 100644
index 00000000000..fb0288b26c8
--- /dev/null
+++ b/cmd/pipelines/test_events_test.go
@@ -0,0 +1,254 @@
+package pipelines
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/url"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/databricks/cli/libs/testserver"
+	databricks "github.com/databricks/databricks-sdk-go"
+	"github.com/databricks/databricks-sdk-go/service/pipelines"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// testEvents builds the events of one test, each with its own event ID.
+type testEvents struct {
+	t *testing.T
+	n int
+}
+
+func newTestEvents(t *testing.T) *testEvents {
+	return &testEvents{t: t}
+}
+
+func (b *testEvents) testCase(id, status, timestamp string, extra ...string) testPipelineEvent {
+	b.n++
+	return decodeTestEvent(b.t, fmt.Sprintf(`{
+		"id": "e%d", "event_type": "test_case_progress", "timestamp": %q,
+		"origin": {"test_case_id": %q},
+		"details": {"test_case_progress": {"status": %q}}%s
+	}`, b.n, timestamp, id, status, strings.Join(append([]string{""}, extra...), ",")))
+}
+
+func (b *testEvents) summary(total, passed, failed, skipped, errored int) testPipelineEvent {
+	b.n++
+	return decodeTestEvent(b.t, fmt.Sprintf(`{
+		"id": "e%d", "event_type": "test_run_summary",
+		"details": {"test_run_summary": {"total": %d, "passed": %d, "failed": %d, "skipped": %d, "errored": %d}}
+	}`, b.n, total, passed, failed, skipped, errored))
+}
+
+func decodeTestEvent(t *testing.T, data string) testPipelineEvent {
+	var e testPipelineEvent
+	require.NoError(t, json.Unmarshal([]byte(data), &e))
+	return e
+}
+
+func TestTestRunAdd(t *testing.T) {
+	ev := newTestEvents(t)
+	run := newTestRun()
+	events := []testPipelineEvent{
+		ev.testCase("tests/test_a.py::test_ok", "PENDING", "2026-01-01T00:00:00.000Z"),
+		ev.testCase("tests/test_a.py::test_ok", "RUNNING", "2026-01-01T00:00:01.000Z"),
+		ev.testCase("tests/test_a.py::test_ok", "PASSED", "2026-01-01T00:00:02.840Z"),
+		ev.testCase("tests/test_a.py::test_bad", "PIPELINE_TEST_CASE_STATUS_FAILED", "2026-01-01T00:00:03.000Z",
+			`"error": {"exceptions": [{"class_name": "AssertionError", "message": "assert 1 == 2"}]}`,
+			`"truncation": {"truncated_fields": [{"field_name": "error"}]}`),
+		// A late RUNNING event does not undo the final outcome.
+		ev.testCase("tests/test_a.py::test_bad", "RUNNING", "2026-01-01T00:00:02.000Z"),
+		ev.summary(2, 1, 1, 0, 0),
+	}
+
+	var finished []string
+	for _, e := range events {
+		if c := run.add(e); c != nil {
+			finished = append(finished, c.id)
+		}
+	}
+	// Reading every page again adds nothing.
+	for _, e := range events {
+		assert.Nil(t, run.add(e))
+	}
+
+	assert.Equal(t, []string{"tests/test_a.py::test_ok", "tests/test_a.py::test_bad"}, finished)
+	require.Len(t, run.cases, 2)
+
+	ok := run.cases[0]
+	assert.Equal(t, testCasePassed, ok.status)
+	d, known := ok.duration()
+	assert.True(t, known)
+	assert.Equal(t, 1840*time.Millisecond, d)
+
+	bad := run.cases[1]
+	assert.Equal(t, testCaseFailed, bad.status)
+	assert.True(t, bad.detailsTruncated)
+	assert.Equal(t, "AssertionError: assert 1 == 2", errorDetailText(bad.errorDetail))
+	d, known = bad.duration()
+	assert.True(t, known)
+	assert.Equal(t, time.Second, d)
+
+	assert.Empty(t, run.incompleteReason(pipelines.UpdateInfoStateFailed))
+}
+
+func TestTestRunIncompleteReason(t *testing.T) {
+	ev := newTestEvents(t)
+	passed := func() testPipelineEvent {
+		return ev.testCase("t::a", "PASSED", "")
+	}
+	tests := []struct {
+		name   string
+		events []testPipelineEvent
+		state  pipelines.UpdateInfoState
+		want   string
+	}{
+		{
+			name:   "no summary after completion",
+			events: []testPipelineEvent{passed()},
+			state:  pipelines.UpdateInfoStateCompleted,
+			want:   "the update ended without a test summary",
+		},
+		{
+			name:   "two summaries",
+			events: []testPipelineEvent{passed(), ev.summary(1, 1, 0, 0, 0), ev.summary(1, 1, 0, 0, 0)},
+			state:  pipelines.UpdateInfoStateCompleted,
+			want:   "received 2 test summaries instead of one",
+		},
+		{
+			name:   "missing outcome",
+			events: []testPipelineEvent{passed(), ev.testCase("t::b", "RUNNING", ""), ev.summary(2, 2, 0, 0, 0)},
+			state:  pipelines.UpdateInfoStateCompleted,
+			want:   "received 1 of 2 test results",
+		},
+		{
+			name:   "case without outcome that the summary leaves out",
+			events: []testPipelineEvent{passed(), ev.testCase("t::b", "RUNNING", ""), ev.summary(1, 1, 0, 0, 0)},
+			state:  pipelines.UpdateInfoStateCompleted,
+			want:   "test t::b has no result",
+		},
+		{
+			name:   "two outcomes for one case",
+			events: []testPipelineEvent{passed(), ev.testCase("t::a", "FAILED", ""), ev.summary(1, 1, 0, 0, 0)},
+			state:  pipelines.UpdateInfoStateCompleted,
+			want:   "test t::a has more than one result",
+		},
+		{
+			name:   "outcomes differ from summary",
+			events: []testPipelineEvent{ev.testCase("t::a", "FAILED", ""), ev.summary(1, 1, 0, 0, 0)},
+			state:  pipelines.UpdateInfoStateCompleted,
+			want:   "the test results do not match the test summary",
+		},
+		{
+			name: "outcome without details",
+			events: []testPipelineEvent{
+				passed(),
+				decodeTestEvent(t, `{"id": "truncated", "event_type": "test_case_progress", "origin": {"test_case_id": "t::b"}}`),
+				ev.summary(1, 1, 0, 0, 0),
+			},
+			state: pipelines.UpdateInfoStateCompleted,
+			want:  "some test results could not be read",
+		},
+		{
+			name:   "unknown status",
+			events: []testPipelineEvent{passed(), ev.testCase("t::b", "PIPELINE_TEST_CASE_STATUS_UNSPECIFIED", ""), ev.summary(1, 1, 0, 0, 0)},
+			state:  pipelines.UpdateInfoStateCompleted,
+			want:   "some test results could not be read",
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			run := newTestRun()
+			for _, e := range tc.events {
+				run.add(e)
+			}
+			assert.Equal(t, tc.want, run.incompleteReason(tc.state))
+		})
+	}
+}
+
+func TestReadTestEventsFollowsPages(t *testing.T) {
+	server := testserver.New(t)
+	var queries []url.Values
+	server.Handle("GET", "/api/2.0/pipelines/{pipeline_id}/events", func(req testserver.Request) any {
+		queries = append(queries, req.URL.Query())
+		if req.URL.Query().Get("page_token") == "" {
+			return map[string]any{"events": []map[string]string{{"id": "e1"}}, "next_page_token": "page2"}
+		}
+		return map[string]any{"events": []map[string]string{{"id": "e2"}}}
+	})
+
+	w, err := databricks.NewWorkspaceClient(&databricks.Config{Host: server.URL, Token: "token"})
+	require.NoError(t, err)
+	events, err := readTestEvents(t.Context(), w, "p1", "u1", "2026-01-01T00:00:01.000Z")
+	require.NoError(t, err)
+
+	var ids []string
+	for _, e := range events {
+		ids = append(ids, e.Id)
+	}
+	assert.Equal(t, []string{"e1", "e2"}, ids)
+	assert.Equal(t, []url.Values{
+		{
+			"filter":      {"update_id = 'u1' AND timestamp >= '2026-01-01T00:00:01.000Z'"},
+			"order_by":    {"timestamp asc"},
+			"max_results": {"250"},
+		},
+		{
+			"page_token":  {"page2"},
+			"max_results": {"250"},
+		},
+	}, queries)
+}
+
+// NFR3: a run of 200 cases spans several pages, and each case is counted once.
+func TestReadTestEvents200Cases(t *testing.T) {
+	const cases = 200
+	var all []map[string]any
+	for i := range cases {
+		id := fmt.Sprintf("/Workspace/p/tests/test_many.py::test_case_%03d", i)
+		for j, status := range []string{"PENDING", "RUNNING", "PASSED"} {
+			all = append(all, map[string]any{
+				"id": fmt.Sprintf("c%d-%d", i, j), "event_type": "test_case_progress", "timestamp": "2026-01-01T00:00:00.000Z",
+				"origin": map[string]any{"test_case_id": id}, "details": map[string]any{"test_case_progress": map[string]any{"status": status}},
+			})
+		}
+	}
+	all = append(all, map[string]any{
+		"id": "summary", "event_type": "test_run_summary",
+		"details": map[string]any{"test_run_summary": map[string]any{"total": cases, "passed": cases}},
+	})
+
+	server := testserver.New(t)
+	pages := 0
+	server.Handle("GET", "/api/2.0/pipelines/{pipeline_id}/events", func(req testserver.Request) any {
+		pages++
+		offset := 0
+		if token := req.URL.Query().Get("page_token"); token != "" {
+			_, err := fmt.Sscan(token, &offset)
+			require.NoError(t, err)
+		}
+		end := min(offset+testEventsPageSize, len(all))
+		page := map[string]any{"events": all[offset:end]}
+		if end < len(all) {
+			page["next_page_token"] = strconv.Itoa(end)
+		}
+		return page
+	})
+
+	w, err := databricks.NewWorkspaceClient(&databricks.Config{Host: server.URL, Token: "token"})
+	require.NoError(t, err)
+	events, err := readTestEvents(t.Context(), w, "p1", "u1", "")
+	require.NoError(t, err)
+	assert.Equal(t, 3, pages)
+
+	run := newTestRun()
+	for _, e := range append(events, events...) {
+		run.add(e)
+	}
+	assert.Empty(t, run.incompleteReason(pipelines.UpdateInfoStateCompleted))
+	assert.Equal(t, testRunSummary{Total: cases, Passed: cases}, run.counts())
+}