From ff81fb70347ae261fa62605262f232d7058864e5 Mon Sep 17 00:00:00 2001
From: Kiran Muddukrishna <kmuddukrishna@squareup.com>
Date: Mon, 17 Aug 2026 16:41:28 +1000
Subject: [PATCH 1/3] Add strategy-wide execution progress tracking

Callers (CLI today, the SchemaBot adapter next) need a machine-readable
view of a running change: phase, sequence step position, retry attempt,
and live pg_stat_progress_create_index counters for concurrent builds.
Native operations leave the copy counters empty so copy-and-swap can
implement the same contract later.
---
 SAFETY.md                               |   2 +-
 docs/architecture.md                    |   1 +
 docs/low-level-design.md                |  14 +-
 pkg/dbconn/dbconn.go                    |  32 ++++
 pkg/dbconn/dbconn_integration_test.go   |   7 +
 pkg/executor/native.go                  |  27 ++++
 pkg/executor/native_integration_test.go |  39 +++++
 pkg/executor/optimistic.go              |  30 +++-
 pkg/executor/sequence.go                |  47 +++++-
 pkg/progress/progress.go                | 197 ++++++++++++++++++++++++
 pkg/progress/progress_test.go           |  75 +++++++++
 11 files changed, 457 insertions(+), 14 deletions(-)
 create mode 100644 pkg/progress/progress.go
 create mode 100644 pkg/progress/progress_test.go

diff --git a/SAFETY.md b/SAFETY.md
index 479c8af..fdcabac 100644
--- a/SAFETY.md
+++ b/SAFETY.md
@@ -31,7 +31,7 @@ The invariant registry (invariant IDs referenced below) lives in
 | `pkg/verdict` — structured outcome contract, rendering, exit codes | ❌ periphery | exists (Phase 1) | — |
 | `pkg/diffplan` — desired schema → routed convergence plan, the declarative front door as a library (the CLI `diff` and embedding orchestrators share it) | ❌ periphery | exists | — |
 | `internal/cli` — CLI, flags, help, prompts | ❌ periphery | `migrate`, `status`, `diff`, `fmt`, `lint`, and `suggest` exist | — |
-| status / progress / advisory rendering, metrics | ❌ periphery | planned | — |
+| `pkg/progress` — strategy-wide native progress snapshots; copy counters reserved for later | ❌ periphery | native progress exists | — |
 | orchestrator adapter | ❌ periphery | planned (Phase 11) | OC-* hold *at* the boundary |
 | `internal/testutil` | ❌ test-only | exists | — |
 
diff --git a/docs/architecture.md b/docs/architecture.md
index 9cb0362..567cf9b 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -136,6 +136,7 @@ different levels of commitment:
 | `pkg/diffplan` | The declarative front door as a library: desired schema in, routed `plan.Report` out — the CLI `diff` and embedding orchestrators share this one pipeline | exists |
 | `pkg/router` | Route classified statements to native / copy-and-swap / refuse dispositions; copy-and-swap reports unavailable until that backend lands | exists (Phase 2.4) |
 | `pkg/executor` | Bounded optimistic native attempt, the concurrent index build, and the autocommit safer-sequence runner, with stable outcome codes; the full `Executor` contract (`Plan`/`Execute`/`Status`/`Abort`) arrives with the copy-and-swap backend | native execution exists |
+| `pkg/progress` | Strategy-wide, pollable progress snapshots: native phase/elapsed time, sequence position, retry attempt, and server-reported concurrent-index work; optional copy counters are reserved for copy-and-swap | native progress exists |
 | `pkg/table` | PK-range chunkers (single-column fast path, composite), dynamic time-based sizing | Phase 4 |
 | `pkg/copier` | Parallel chunked copy into the shadow table (never overwrites) | Phase 4 |
 | `pkg/checksum` | The mandatory correctness gate; continuous checker; repair primitive | Phase 5 |
diff --git a/docs/low-level-design.md b/docs/low-level-design.md
index cc70af9..7a9129f 100644
--- a/docs/low-level-design.md
+++ b/docs/low-level-design.md
@@ -811,12 +811,14 @@ At the library seam, each executor outcome maps to a stable string code
 embedding `pkg/executor` branches on one vocabulary; the CLI's verdict JSON carries the same
 codes — an execution failure ends in a `failed` verdict (exit 1, distinct from the refusal
 exit 2) with the code, the failed step, and the committed prefix in `executed_sql`, so
-automation can distinguish nothing-committed from partial state left behind. Remaining
-Phase 3 work, roughly in order:
-
-- bound lock acquisition with timeout and retry for the blocking idioms,
-- progress reporting (`pg_stat_progress_create_index` by the build's backend PID, which the
-  executor already captures for its ownership proof).
+automation can distinguish nothing-committed from partial state left behind. Native execution
+exposes a caller-owned `progress.Tracker`. Embedders run a blocking executor
+call in their own bounded task and poll `Tracker.Progress(ctx)`: sequence position and elapsed
+time come from in-process state, while an active concurrent index build is read on demand from
+`pg_stat_progress_create_index` by the build's backend PID. There is no background poller to
+own or stop, and a missing progress-view row is represented as an inactive observation rather
+than an error. The same snapshot already reserves optional row and byte copy counters for the
+copy-and-swap backend.
 
 The copy-and-swap backend, including change capture, copying, applying, checksumming, and
 cutover, follows Phase 3.
diff --git a/pkg/dbconn/dbconn.go b/pkg/dbconn/dbconn.go
index 915785d..6dd1508 100644
--- a/pkg/dbconn/dbconn.go
+++ b/pkg/dbconn/dbconn.go
@@ -8,6 +8,7 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
+	"errors"
 	"fmt"
 	"log/slog"
 	"os"
@@ -116,6 +117,37 @@ func ServerMajor(ctx context.Context, pool *pgxpool.Pool) (int, error) {
 	return major, nil
 }
 
+// IndexBuildProgress is one server observation of a concurrent index build.
+type IndexBuildProgress struct {
+	Phase       string
+	BlocksDone  uint64
+	BlocksTotal uint64
+	TuplesDone  uint64
+	TuplesTotal uint64
+}
+
+// RowQuerier is the session capability needed for a progress observation.
+type RowQuerier interface {
+	QueryRow(context.Context, string, ...any) pgx.Row
+}
+
+// ConcurrentIndexProgress reads the active build owned by backendPID. The
+// boolean is false when PostgreSQL has not published the row yet or the build
+// has already left the progress view.
+func ConcurrentIndexProgress(ctx context.Context, session RowQuerier, backendPID uint32) (IndexBuildProgress, bool, error) {
+	var p IndexBuildProgress
+	err := session.QueryRow(ctx, `SELECT phase, blocks_done, blocks_total, tuples_done, tuples_total
+		FROM pg_catalog.pg_stat_progress_create_index WHERE pid = $1`, backendPID).
+		Scan(&p.Phase, &p.BlocksDone, &p.BlocksTotal, &p.TuplesDone, &p.TuplesTotal)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return p, false, nil
+	}
+	if err != nil {
+		return p, false, fmt.Errorf("read concurrent index progress for backend %d: %w", backendPID, err)
+	}
+	return p, true, nil
+}
+
 // buildPoolConfig translates Config into a pgxpool configuration. It is pure
 // (no dialing), so every option's wiring is unit-testable without a server.
 func buildPoolConfig(cfg Config) (*pgxpool.Config, error) {
diff --git a/pkg/dbconn/dbconn_integration_test.go b/pkg/dbconn/dbconn_integration_test.go
index c371df5..1c4db04 100644
--- a/pkg/dbconn/dbconn_integration_test.go
+++ b/pkg/dbconn/dbconn_integration_test.go
@@ -40,6 +40,13 @@ func TestPoolIntegration(t *testing.T) {
 		assert.Equal(t, want, version)
 	})
 
+	t.Run("absent concurrent index progress is not an error", func(t *testing.T) {
+		observation, active, err := dbconn.ConcurrentIndexProgress(t.Context(), pool, 0)
+		require.NoError(t, err)
+		assert.False(t, active)
+		assert.Empty(t, observation.Phase)
+	})
+
 	t.Run("statement_timeout cancels runaway work", func(t *testing.T) {
 		_, err := pool.Exec(t.Context(), "SELECT pg_sleep(5)")
 		require.Error(t, err)
diff --git a/pkg/executor/native.go b/pkg/executor/native.go
index bf69dc2..69726c4 100644
--- a/pkg/executor/native.go
+++ b/pkg/executor/native.go
@@ -24,6 +24,7 @@ import (
 	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/jackc/pgx/v5/pgxpool"
 
+	"github.com/block/pg-sprite/pkg/progress"
 	"github.com/block/pg-sprite/pkg/statement"
 )
 
@@ -266,6 +267,22 @@ func (e *InvalidIndexError) Unwrap() []error {
 // whole budget, while a concurrent build takes only SHARE UPDATE EXCLUSIVE —
 // long builds on large tables are its purpose.
 func BuildIndexConcurrently(ctx context.Context, pool *pgxpool.Pool, sql string, b ConcurrentBudget) (IndexBuildReport, error) {
+	return buildIndexConcurrently(ctx, pool, sql, b, nil)
+}
+
+// BuildIndexConcurrentlyWithProgress runs a concurrent build while updating
+// tracker. The caller may poll tracker concurrently with this blocking call.
+func BuildIndexConcurrentlyWithProgress(ctx context.Context, pool *pgxpool.Pool, sql string, b ConcurrentBudget, tracker *progress.Tracker) (rep IndexBuildReport, err error) {
+	if tracker == nil {
+		return rep, fmt.Errorf("progress tracker is required")
+	}
+	tracker.Start(1, progress.OperationConcurrentIndex)
+	tracker.StartStep(1, progress.OperationConcurrentIndex)
+	defer func() { tracker.Finish(err) }()
+	return buildIndexConcurrently(ctx, pool, sql, b, tracker)
+}
+
+func buildIndexConcurrently(ctx context.Context, pool *pgxpool.Pool, sql string, b ConcurrentBudget, tracker *progress.Tracker) (IndexBuildReport, error) {
 	var rep IndexBuildReport
 	if err := b.validate(); err != nil {
 		return rep, err
@@ -323,10 +340,20 @@ func BuildIndexConcurrently(ctx context.Context, pool *pgxpool.Pool, sql string,
 	// The backend PID anchors the post-failure ownership proof: recovery
 	// waits for this backend to stop before trusting the catalog.
 	pid := conn.Conn().PgConn().PID()
+	if tracker != nil {
+		tracker.SetConcurrentBuild(verdictConn, pid)
+	}
 
 	start := time.Now()
+	if tracker != nil {
+		start = tracker.Now()
+	}
 	_, buildErr := conn.Exec(ctx, sql)
 	elapsed := time.Since(start)
+	if tracker != nil {
+		elapsed = tracker.Now().Sub(start)
+		tracker.StopConcurrentBuild()
+	}
 	if buildErr == nil {
 		return verifiedBuildReport(ctx, conn, build, target, elapsed)
 	}
diff --git a/pkg/executor/native_integration_test.go b/pkg/executor/native_integration_test.go
index 887986c..8143996 100644
--- a/pkg/executor/native_integration_test.go
+++ b/pkg/executor/native_integration_test.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 	"testing"
 	"time"
 
@@ -16,6 +17,7 @@ import (
 	"github.com/block/pg-sprite/internal/testutil"
 	"github.com/block/pg-sprite/pkg/dbconn"
 	"github.com/block/pg-sprite/pkg/executor"
+	"github.com/block/pg-sprite/pkg/progress"
 )
 
 // buildBudget bounds test builds: generous enough that a healthy build on a
@@ -83,6 +85,43 @@ func TestBuildIndexConcurrentlyBuildsValidIndex(t *testing.T) {
 	assert.True(t, valid, "the index must be valid")
 }
 
+func TestBuildIndexConcurrentlyReportsServerProgressAndFinishes(t *testing.T) {
+	pool, schema := newPool(t)
+	_, err := pool.Exec(t.Context(), fmt.Sprintf(`CREATE TABLE %s.progress_t AS
+		SELECT n AS id, repeat(md5(n::text), 4) AS payload FROM generate_series(1, 1000000) n`, schema))
+	require.NoError(t, err)
+	tracker, err := progress.NewTracker(progress.WallClock{})
+	require.NoError(t, err)
+
+	type result struct{ err error }
+	results := make(chan result, 1)
+	var workers sync.WaitGroup
+	workers.Go(func() {
+		_, buildErr := executor.BuildIndexConcurrentlyWithProgress(t.Context(), pool,
+			fmt.Sprintf("CREATE INDEX CONCURRENTLY progress_idx ON %s.progress_t (payload)", schema), buildBudget, tracker)
+		results <- result{err: buildErr}
+	})
+	t.Cleanup(workers.Wait)
+
+	var observed progress.Snapshot
+	require.Eventually(t, func() bool {
+		var progressErr error
+		observed, progressErr = tracker.Progress(t.Context())
+		return progressErr == nil && observed.Detail.ServerPhase != ""
+	}, 30*time.Second, 10*time.Millisecond, "the active build must publish server progress")
+	require.NotNil(t, observed.Detail.Work)
+	assert.LessOrEqual(t, observed.Detail.Work.BlocksDone, observed.Detail.Work.BlocksTotal)
+	assert.LessOrEqual(t, observed.Detail.Work.TuplesDone, observed.Detail.Work.TuplesTotal)
+
+	buildResult := <-results
+	require.NoError(t, buildResult.err)
+	workers.Wait()
+	finished, err := tracker.Progress(t.Context())
+	require.NoError(t, err)
+	assert.Equal(t, progress.PhaseFinished, finished.Phase)
+	assert.False(t, finished.Detail.Active, "a completed build has no active progress row")
+}
+
 // TestBuildIndexConcurrentlyRefusesSingleConnectionPool covers the
 // admission-time pool guard: the verdict session is a correctness
 // dependency reserved alongside the build session, so a pool that cannot
diff --git a/pkg/executor/optimistic.go b/pkg/executor/optimistic.go
index b92e5ed..bfeff1c 100644
--- a/pkg/executor/optimistic.go
+++ b/pkg/executor/optimistic.go
@@ -26,6 +26,7 @@ import (
 	"github.com/jackc/pgx/v5/pgxpool"
 
 	"github.com/block/pg-sprite/pkg/preflight"
+	"github.com/block/pg-sprite/pkg/progress"
 	"github.com/block/pg-sprite/pkg/statement"
 )
 
@@ -179,6 +180,22 @@ func (b Budget) validate() error {
 // work that exceeded its execution budget is not a lock-acquisition
 // strategy.
 func ExecuteNative(ctx context.Context, pool *pgxpool.Pool, pt preflight.PreflightedTable, st statement.Statement, b Budget, retry RetryPolicy) error {
+	return executeNative(ctx, pool, pt, st, b, retry, nil)
+}
+
+// ExecuteNativeWithProgress runs an optimistic native attempt while updating
+// tracker. The caller may poll tracker concurrently with this blocking call.
+func ExecuteNativeWithProgress(ctx context.Context, pool *pgxpool.Pool, pt preflight.PreflightedTable, st statement.Statement, b Budget, retry RetryPolicy, tracker *progress.Tracker) (err error) {
+	if tracker == nil {
+		return fmt.Errorf("progress tracker is required")
+	}
+	tracker.Start(1, progress.OperationOptimistic)
+	tracker.StartStep(1, progress.OperationOptimistic)
+	defer func() { tracker.Finish(err) }()
+	return executeNative(ctx, pool, pt, st, b, retry, tracker)
+}
+
+func executeNative(ctx context.Context, pool *pgxpool.Pool, pt preflight.PreflightedTable, st statement.Statement, b Budget, retry RetryPolicy, tracker *progress.Tracker) error {
 	if err := b.validate(); err != nil {
 		return err
 	}
@@ -191,9 +208,13 @@ func ExecuteNative(ctx context.Context, pool *pgxpool.Pool, pt preflight.Preflig
 		return fmt.Errorf("%w: ST-7: statement targets %q but preflight verified %q",
 			ErrInvariantViolation, qualifiedName(st.Schema(), st.Table()), qualifiedName(pt.Schema(), pt.Table()))
 	}
-	return executeWithLockRetry(ctx, retry, func(ctx context.Context) error {
+	return executeWithLockRetryObserved(ctx, retry, func(ctx context.Context) error {
 		return executeNativeAttempt(ctx, pool, st, b)
-	}, sleepContext)
+	}, sleepContext, func(attempt int) {
+		if tracker != nil {
+			tracker.SetAttempt(attempt)
+		}
+	})
 }
 
 func executeNativeAttempt(ctx context.Context, pool *pgxpool.Pool, st statement.Statement, b Budget) error {
@@ -234,7 +255,12 @@ func executeNativeAttempt(ctx context.Context, pool *pgxpool.Pool, st statement.
 type sleepFunc func(context.Context, time.Duration) error
 
 func executeWithLockRetry(ctx context.Context, policy RetryPolicy, attempt func(context.Context) error, sleep sleepFunc) error {
+	return executeWithLockRetryObserved(ctx, policy, attempt, sleep, func(int) {})
+}
+
+func executeWithLockRetryObserved(ctx context.Context, policy RetryPolicy, attempt func(context.Context) error, sleep sleepFunc, observe func(int)) error {
 	for i := 1; i <= policy.MaxAttempts; i++ {
+		observe(i)
 		err := attempt(ctx)
 		if err == nil {
 			return nil
diff --git a/pkg/executor/sequence.go b/pkg/executor/sequence.go
index f8aea86..6e5f59a 100644
--- a/pkg/executor/sequence.go
+++ b/pkg/executor/sequence.go
@@ -33,6 +33,7 @@ import (
 	"github.com/jackc/pgx/v5/pgxpool"
 
 	"github.com/block/pg-sprite/pkg/preflight"
+	"github.com/block/pg-sprite/pkg/progress"
 	"github.com/block/pg-sprite/pkg/statement"
 )
 
@@ -222,6 +223,21 @@ type sequenceStep struct {
 // lock_timeout retries on each owner-gated step, exactly as in
 // ExecuteNative.
 func RunSequence(ctx context.Context, pool *pgxpool.Pool, pt preflight.PreflightedTable, steps []string, b SequenceBudget, retry RetryPolicy) (SequenceReport, error) {
+	return runSequence(ctx, pool, pt, steps, b, retry, nil)
+}
+
+// RunSequenceWithProgress runs a sequence while updating tracker with the
+// current step and its execution class. The caller may poll concurrently.
+func RunSequenceWithProgress(ctx context.Context, pool *pgxpool.Pool, pt preflight.PreflightedTable, steps []string, b SequenceBudget, retry RetryPolicy, tracker *progress.Tracker) (rep SequenceReport, err error) {
+	if tracker == nil {
+		return rep, fmt.Errorf("progress tracker is required")
+	}
+	tracker.Start(len(steps), progress.OperationBrief)
+	defer func() { tracker.Finish(err) }()
+	return runSequence(ctx, pool, pt, steps, b, retry, tracker)
+}
+
+func runSequence(ctx context.Context, pool *pgxpool.Pool, pt preflight.PreflightedTable, steps []string, b SequenceBudget, retry RetryPolicy, tracker *progress.Tracker) (SequenceReport, error) {
 	var rep SequenceReport
 	if err := b.validate(); err != nil {
 		return rep, err
@@ -260,22 +276,26 @@ func RunSequence(ctx context.Context, pool *pgxpool.Pool, pt preflight.Preflight
 	}
 	for i, step := range admitted {
 		start := time.Now()
+		if tracker != nil {
+			tracker.StartStep(i+1, progressOperation(step.kind))
+			start = tracker.Now()
+		}
 		var indexReport *IndexBuildReport
 		switch step.kind {
 		case StepConcurrentIndexBuild:
-			r, buildErr := BuildIndexConcurrently(ctx, pool, step.st.SQL(), b.Concurrent)
+			r, buildErr := buildIndexConcurrently(ctx, pool, step.st.SQL(), b.Concurrent, tracker)
 			err = buildErr
 			if buildErr == nil {
 				indexReport = &r
 			}
 		case StepValidateConstraint:
-			err = ExecuteNative(ctx, pool, pt, step.st, Budget{
+			err = executeNative(ctx, pool, pt, step.st, Budget{
 				LockTimeout:      b.Validate.LockTimeout,
 				StatementTimeout: b.Validate.Overall,
-			}, retry)
+			}, retry, tracker)
 			err = corroborateValidateCancel(err, b.Validate, time.Since(start))
 		case StepBrief:
-			err = ExecuteNative(ctx, pool, pt, step.st, b.Brief, retry)
+			err = executeNative(ctx, pool, pt, step.st, b.Brief, retry, tracker)
 		default:
 			// Admission produces only the three kinds above; an unknown
 			// kind here is a programming error and aborts fail-closed.
@@ -284,10 +304,14 @@ func RunSequence(ctx context.Context, pool *pgxpool.Pool, pt preflight.Preflight
 		if err != nil {
 			return rep, &SequenceStepError{Step: i + 1, Total: len(admitted), Kind: step.kind, SQL: step.st.SQL(), Err: err}
 		}
+		duration := time.Since(start)
+		if tracker != nil {
+			duration = tracker.Now().Sub(start)
+		}
 		rep.Steps = append(rep.Steps, StepReport{
 			SQL:      step.st.SQL(),
 			Kind:     step.kind,
-			Duration: time.Since(start),
+			Duration: duration,
 			Index:    indexReport,
 		})
 	}
@@ -307,6 +331,19 @@ func sequenceTargetFacts(ctx context.Context, pool *pgxpool.Pool, schema, table
 	return facts.Partitioned(), facts.ServerMajor(), nil
 }
 
+func progressOperation(kind StepKind) progress.Operation {
+	switch kind {
+	case StepBrief:
+		return progress.OperationBrief
+	case StepValidateConstraint:
+		return progress.OperationValidate
+	case StepConcurrentIndexBuild:
+		return progress.OperationConcurrentIndex
+	default:
+		return ""
+	}
+}
+
 // sequenceHasConcurrentBuild reports whether any admitted step is a
 // concurrent index build — the class whose executor needs the two-connection
 // pool guarantee.
diff --git a/pkg/progress/progress.go b/pkg/progress/progress.go
new file mode 100644
index 0000000..e793d91
--- /dev/null
+++ b/pkg/progress/progress.go
@@ -0,0 +1,197 @@
+// Package progress defines the strategy-wide, machine-readable execution
+// progress contract. It deliberately contains copy counters that native
+// operations leave empty so copy-and-swap can implement the same contract.
+package progress
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/block/pg-sprite/pkg/dbconn"
+)
+
+// Clock supplies time to progress state so core executors remain deterministic.
+type Clock interface {
+	Now() time.Time
+}
+
+// WallClock reads the process wall clock.
+type WallClock struct{}
+
+// Now returns the current wall-clock time.
+func (WallClock) Now() time.Time { return time.Now() }
+
+// Phase is the overall execution phase.
+type Phase string
+
+const (
+	// PhasePending means execution has not started.
+	PhasePending Phase = "pending"
+	// PhaseRunning means execution is active.
+	PhaseRunning Phase = "running"
+	// PhaseFinished means execution completed successfully.
+	PhaseFinished Phase = "finished"
+	// PhaseFailed means execution reached a terminal failure.
+	PhaseFailed Phase = "failed"
+)
+
+// Operation is the current operation's execution class.
+type Operation string
+
+const (
+	// OperationOptimistic is one bounded direct native attempt.
+	OperationOptimistic Operation = "optimistic"
+	// OperationBrief is a brief transactional sequence step.
+	OperationBrief Operation = "brief"
+	// OperationValidate is a constraint-validation scan.
+	OperationValidate Operation = "validate-constraint"
+	// OperationConcurrentIndex is a concurrent index build.
+	OperationConcurrentIndex Operation = "concurrent-index-build"
+)
+
+// Work reports optional server-observed work. Rows and bytes are reserved for
+// copy-and-swap; native operations do not fabricate them.
+type Work struct {
+	RowsCopied  uint64 `json:"rows_copied,omitempty"`
+	RowsTotal   uint64 `json:"rows_total,omitempty"`
+	BytesCopied uint64 `json:"bytes_copied,omitempty"`
+	BytesTotal  uint64 `json:"bytes_total,omitempty"`
+	BlocksDone  uint64 `json:"blocks_done,omitempty"`
+	BlocksTotal uint64 `json:"blocks_total,omitempty"`
+	TuplesDone  uint64 `json:"tuples_done,omitempty"`
+	TuplesTotal uint64 `json:"tuples_total,omitempty"`
+}
+
+// Detail describes the operation currently executing.
+type Detail struct {
+	Operation   Operation `json:"operation,omitempty"`
+	ServerPhase string    `json:"server_phase,omitempty"`
+	Active      bool      `json:"active"`
+	Attempt     int       `json:"attempt,omitempty"`
+	Work        *Work     `json:"work,omitempty"`
+}
+
+// Snapshot is one immutable progress observation.
+type Snapshot struct {
+	Phase       Phase         `json:"phase"`
+	Step        int           `json:"step,omitempty"`
+	TotalSteps  int           `json:"total_steps,omitempty"`
+	Elapsed     time.Duration `json:"elapsed_ns"`
+	StepElapsed time.Duration `json:"step_elapsed_ns,omitempty"`
+	Detail      Detail        `json:"detail"`
+}
+
+// Tracker is a concurrency-safe progress source. The caller owns it; it has
+// no goroutines. Progress performs the one read needed for an active index
+// build, making polling lifetime identical to the caller's context.
+type Tracker struct {
+	mu        sync.RWMutex
+	clock     Clock
+	session   dbconn.RowQuerier
+	phase     Phase
+	started   time.Time
+	stepStart time.Time
+	step      int
+	total     int
+	detail    Detail
+	buildPID  uint32
+}
+
+// NewTracker constructs an idle tracker using clock.
+func NewTracker(clock Clock) (*Tracker, error) {
+	if clock == nil {
+		return nil, fmt.Errorf("progress clock is required")
+	}
+	return &Tracker{clock: clock, phase: PhasePending}, nil
+}
+
+// Now returns the tracker's injected time for executor duration accounting.
+func (t *Tracker) Now() time.Time { return t.clock.Now() }
+
+// Start records the beginning of an execution.
+func (t *Tracker) Start(total int, operation Operation) {
+	now := t.clock.Now()
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.phase, t.started, t.stepStart = PhaseRunning, now, now
+	t.total, t.detail = total, Detail{Operation: operation, Active: true}
+}
+
+// StartStep advances a sequence to a 1-based step.
+func (t *Tracker) StartStep(step int, operation Operation) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.step, t.stepStart = step, t.clock.Now()
+	t.detail = Detail{Operation: operation, Active: true}
+	t.buildPID = 0
+}
+
+// SetAttempt records the current bounded retry attempt.
+func (t *Tracker) SetAttempt(attempt int) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.detail.Attempt = attempt
+}
+
+// SetConcurrentBuild enables on-demand server progress for pid. The executor
+// supplies its reserved verdict session so polling cannot starve behind the
+// build session even when the pool has only two connections.
+func (t *Tracker) SetConcurrentBuild(session dbconn.RowQuerier, pid uint32) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.session, t.buildPID = session, pid
+}
+
+// StopConcurrentBuild waits for an in-flight observation and releases the
+// reserved session back to the executor before its catalog verdict.
+func (t *Tracker) StopConcurrentBuild() {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.buildPID = 0
+}
+
+// Finish records a terminal execution outcome.
+func (t *Tracker) Finish(err error) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if err == nil {
+		t.phase = PhaseFinished
+	} else {
+		t.phase = PhaseFailed
+	}
+	t.detail.Active = false
+	t.buildPID = 0
+}
+
+// Progress returns a snapshot and, for an active concurrent index build,
+// queries PostgreSQL's progress view by the executor-owned backend PID.
+func (t *Tracker) Progress(ctx context.Context) (Snapshot, error) {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+	now := t.clock.Now()
+	s := Snapshot{Phase: t.phase, Step: t.step, TotalSteps: t.total, Detail: t.detail}
+	if !t.started.IsZero() {
+		s.Elapsed = now.Sub(t.started)
+		s.StepElapsed = now.Sub(t.stepStart)
+	}
+	session, pid := t.session, t.buildPID
+	if pid == 0 || session == nil || s.Phase != PhaseRunning {
+		return s, nil
+	}
+	p, active, err := dbconn.ConcurrentIndexProgress(ctx, session, pid)
+	if err != nil {
+		return Snapshot{}, err
+	}
+	if !active {
+		s.Detail.Active = false
+		return s, nil
+	}
+	work := Work{
+		BlocksDone: p.BlocksDone, BlocksTotal: p.BlocksTotal,
+		TuplesDone: p.TuplesDone, TuplesTotal: p.TuplesTotal,
+	}
+	s.Detail.Active, s.Detail.ServerPhase, s.Detail.Work = true, p.Phase, &work
+	return s, nil
+}
diff --git a/pkg/progress/progress_test.go b/pkg/progress/progress_test.go
new file mode 100644
index 0000000..1af2faf
--- /dev/null
+++ b/pkg/progress/progress_test.go
@@ -0,0 +1,75 @@
+package progress_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/block/pg-sprite/pkg/progress"
+)
+
+type fakeClock struct{ now time.Time }
+
+func (c *fakeClock) Now() time.Time { return c.now }
+
+func TestTrackerReportsSequencePositionAndInjectedElapsed(t *testing.T) {
+	clock := &fakeClock{now: time.Unix(100, 0)}
+	tracker, err := progress.NewTracker(clock)
+	require.NoError(t, err)
+
+	tracker.Start(3, progress.OperationBrief)
+	clock.now = clock.now.Add(2 * time.Second)
+	tracker.StartStep(2, progress.OperationValidate)
+	tracker.SetAttempt(2)
+	clock.now = clock.now.Add(750 * time.Millisecond)
+
+	snapshot, err := tracker.Progress(t.Context())
+	require.NoError(t, err)
+	assert.Equal(t, progress.PhaseRunning, snapshot.Phase)
+	assert.Equal(t, 2, snapshot.Step)
+	assert.Equal(t, 3, snapshot.TotalSteps)
+	assert.Equal(t, 2750*time.Millisecond, snapshot.Elapsed)
+	assert.Equal(t, 750*time.Millisecond, snapshot.StepElapsed)
+	assert.Equal(t, progress.OperationValidate, snapshot.Detail.Operation)
+	assert.Equal(t, 2, snapshot.Detail.Attempt)
+	assert.Nil(t, snapshot.Detail.Work, "native progress must not fabricate copy counters")
+}
+
+func TestTrackerSequenceStepsAdvanceMonotonically(t *testing.T) {
+	clock := &fakeClock{now: time.Unix(100, 0)}
+	tracker, err := progress.NewTracker(clock)
+	require.NoError(t, err)
+	tracker.Start(3, progress.OperationBrief)
+
+	for step := 1; step <= 3; step++ {
+		tracker.StartStep(step, progress.OperationBrief)
+		snapshot, progressErr := tracker.Progress(t.Context())
+		require.NoError(t, progressErr)
+		assert.Equal(t, step, snapshot.Step)
+		assert.Equal(t, 3, snapshot.TotalSteps)
+	}
+	tracker.Finish(nil)
+	terminal, err := tracker.Progress(t.Context())
+	require.NoError(t, err)
+	assert.Equal(t, progress.PhaseFinished, terminal.Phase)
+	assert.Equal(t, 3, terminal.Step)
+}
+
+func TestTrackerReportsTerminalState(t *testing.T) {
+	tracker, err := progress.NewTracker(&fakeClock{now: time.Unix(100, 0)})
+	require.NoError(t, err)
+	tracker.Start(1, progress.OperationOptimistic)
+	tracker.Finish(nil)
+
+	snapshot, err := tracker.Progress(t.Context())
+	require.NoError(t, err)
+	assert.Equal(t, progress.PhaseFinished, snapshot.Phase)
+	assert.False(t, snapshot.Detail.Active)
+}
+
+func TestNewTrackerRequiresClock(t *testing.T) {
+	_, err := progress.NewTracker(nil)
+	require.Error(t, err)
+}

From c4fe274b90b689e2fdb2b501bad1e9c6ccaeca62 Mon Sep 17 00:00:00 2001
From: Kiran Muddukrishna <kmuddukrishna@squareup.com>
Date: Mon, 17 Aug 2026 19:50:44 +1000
Subject: [PATCH 2/3] Address PR review: serialize pollers, unify clocks, cover
 WithProgress API

Split the tracker's one RWMutex into a memory-state lock and a poll lock:
concurrent Progress() calls previously shared the reserved pgx connection
under RLock (a pgx.Conn is not safe for concurrent use), and a slow poll
could gate the executor's own state updates. Budget corroboration now reads
the same injected clock that produced its start instant, matching the step
report. Adds the missing test coverage for all three *WithProgress entry
points, the retry-attempt observer wiring, the server-progress merge
branches, and the failing-build-under-polling session handoff.
---
 SAFETY.md                                   |   4 +
 pkg/executor/native.go                      |   3 +-
 pkg/executor/native_integration_test.go     |  47 +++++
 pkg/executor/optimistic_integration_test.go |  60 ++++++
 pkg/executor/retry_internal_test.go         |  34 ++++
 pkg/executor/sequence.go                    |  18 +-
 pkg/executor/sequence_integration_test.go   |  60 ++++++
 pkg/executor/sequence_internal_test.go      |  29 +++
 pkg/executor/withprogress_test.go           |  36 ++++
 pkg/progress/progress.go                    |  23 ++-
 pkg/progress/progress_test.go               | 208 ++++++++++++++++++++
 11 files changed, 511 insertions(+), 11 deletions(-)
 create mode 100644 pkg/executor/withprogress_test.go

diff --git a/SAFETY.md b/SAFETY.md
index fdcabac..aa6fe36 100644
--- a/SAFETY.md
+++ b/SAFETY.md
@@ -62,6 +62,10 @@ The short version — the full rules live in [docs/tcb-model.md](docs/tcb-model.
   `pgx/v5`, the parse boundary (`pkg/statement` → `wasilibs/go-pgquery`, the real PostgreSQL
   grammar — the native executor re-verifies statement shape itself rather than trusting the
   caller's classification; the grammar is load-bearing expertise, not copyable mechanics),
+  `pkg/progress` (the executors' progress-observation seam: they write state into a
+  caller-owned tracker whose mutators take only a memory lock, and its polling reads ride
+  the reserved verdict session behind a separate poll lock — so a slow or hung observation
+  can never gate the executor's own state updates),
   stdlib. The future decode path will add `pglogrepl`. Adding one requires a recorded decision (see the rubric in
   [docs/tcb-model.md](docs/tcb-model.md) — copy small things, take pinned dependencies only
   for load-bearing expertise).
diff --git a/pkg/executor/native.go b/pkg/executor/native.go
index 69726c4..3889992 100644
--- a/pkg/executor/native.go
+++ b/pkg/executor/native.go
@@ -349,9 +349,8 @@ func buildIndexConcurrently(ctx context.Context, pool *pgxpool.Pool, sql string,
 		start = tracker.Now()
 	}
 	_, buildErr := conn.Exec(ctx, sql)
-	elapsed := time.Since(start)
+	elapsed := elapsedSince(tracker, start)
 	if tracker != nil {
-		elapsed = tracker.Now().Sub(start)
 		tracker.StopConcurrentBuild()
 	}
 	if buildErr == nil {
diff --git a/pkg/executor/native_integration_test.go b/pkg/executor/native_integration_test.go
index 8143996..b0b6538 100644
--- a/pkg/executor/native_integration_test.go
+++ b/pkg/executor/native_integration_test.go
@@ -122,6 +122,53 @@ func TestBuildIndexConcurrentlyReportsServerProgressAndFinishes(t *testing.T) {
 	assert.False(t, finished.Detail.Active, "a completed build has no active progress row")
 }
 
+// TestBuildIndexConcurrentlyWithProgressFailingBuildUnderPolling covers the
+// reserved-session handoff on the failure path: a poller hammers Progress
+// for the build's entire life while the build fails, and the executor must
+// still get exclusive use of the verdict session for its catalog verdict.
+// A regression that weakens StopConcurrentBuild's drain surfaces here as a
+// wire-protocol error on the shared connection or a race-detector report.
+func TestBuildIndexConcurrentlyWithProgressFailingBuildUnderPolling(t *testing.T) {
+	pool, schema := newPool(t)
+	// Duplicates guarantee the unique build fails after creating its
+	// catalog entry; the row count gives the poller a window to overlap
+	// the build and its failure verdict.
+	_, err := pool.Exec(t.Context(), fmt.Sprintf(`CREATE TABLE %s.t AS
+		SELECT n AS id, n %% 1000 AS c FROM generate_series(1, 100000) n`, schema))
+	require.NoError(t, err)
+	tracker, err := progress.NewTracker(progress.WallClock{})
+	require.NoError(t, err)
+
+	stop := make(chan struct{})
+	var pollers sync.WaitGroup
+	pollers.Go(func() {
+		for {
+			select {
+			case <-stop:
+				return
+			default:
+			}
+			_, pollErr := tracker.Progress(t.Context())
+			assert.NoError(t, pollErr, "polling must stay clean while the build fails")
+		}
+	})
+
+	_, buildErr := executor.BuildIndexConcurrentlyWithProgress(t.Context(), pool,
+		fmt.Sprintf("CREATE UNIQUE INDEX CONCURRENTLY idx_dup ON %s.t (c)", schema), buildBudget, tracker)
+	close(stop)
+	pollers.Wait()
+
+	require.ErrorIs(t, buildErr, executor.ErrBuildLeftInvalidIndex,
+		"the failure verdict must be reached despite concurrent polling")
+	var invalidErr *executor.InvalidIndexError
+	require.ErrorAs(t, buildErr, &invalidErr)
+	assert.Equal(t, "idx_dup", invalidErr.Index)
+	snapshot, err := tracker.Progress(t.Context())
+	require.NoError(t, err)
+	assert.Equal(t, progress.PhaseFailed, snapshot.Phase)
+	assert.False(t, snapshot.Detail.Active)
+}
+
 // TestBuildIndexConcurrentlyRefusesSingleConnectionPool covers the
 // admission-time pool guard: the verdict session is a correctness
 // dependency reserved alongside the build session, so a pool that cannot
diff --git a/pkg/executor/optimistic_integration_test.go b/pkg/executor/optimistic_integration_test.go
index bce576a..3f14efc 100644
--- a/pkg/executor/optimistic_integration_test.go
+++ b/pkg/executor/optimistic_integration_test.go
@@ -14,6 +14,7 @@ import (
 	"github.com/block/pg-sprite/pkg/dbconn"
 	"github.com/block/pg-sprite/pkg/executor"
 	"github.com/block/pg-sprite/pkg/preflight"
+	"github.com/block/pg-sprite/pkg/progress"
 	"github.com/block/pg-sprite/pkg/statement"
 )
 
@@ -136,6 +137,65 @@ func TestExecuteNativeSurfacesOperationalErrors(t *testing.T) {
 	assert.NotErrorAs(t, err, &budgetErr)
 }
 
+func TestExecuteNativeWithProgressCommitsAndFinishes(t *testing.T) {
+	pool, schema := newPool(t)
+	_, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY)", schema))
+	require.NoError(t, err)
+	pt := mustPreflight(t, pool, schema, "t")
+	tracker, err := progress.NewTracker(progress.WallClock{})
+	require.NoError(t, err)
+
+	st := mustParse(t, fmt.Sprintf("ALTER TABLE %s.t ADD COLUMN age int NOT NULL DEFAULT 0", schema))
+	require.NoError(t, executor.ExecuteNativeWithProgress(t.Context(), pool, pt, st, budget,
+		executor.DefaultRetryPolicy(), tracker))
+
+	assert.Equal(t, "integer", columnType(t, pool, schema, "t", "age"), "the committed change must be visible")
+	snapshot, err := tracker.Progress(t.Context())
+	require.NoError(t, err)
+	assert.Equal(t, progress.PhaseFinished, snapshot.Phase)
+	assert.Equal(t, 1, snapshot.Step)
+	assert.Equal(t, 1, snapshot.TotalSteps)
+	assert.Equal(t, progress.OperationOptimistic, snapshot.Detail.Operation)
+	assert.Equal(t, 1, snapshot.Detail.Attempt, "the one successful attempt must be observed")
+	assert.False(t, snapshot.Detail.Active)
+}
+
+// A blocked attempt exhausts its bounded retries; the tracker must report
+// every retry attempt as it runs and a failed terminal phase at the end.
+func TestExecuteNativeWithProgressReportsRetriesAndFailure(t *testing.T) {
+	pool, schema := newPool(t)
+	_, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY)", schema))
+	require.NoError(t, err)
+	pt := mustPreflight(t, pool, schema, "t")
+	tracker, err := progress.NewTracker(progress.WallClock{})
+	require.NoError(t, err)
+
+	// A second session holds ACCESS EXCLUSIVE for the whole test, so the
+	// attempt can never be granted its lock.
+	blocker, err := pool.Begin(t.Context())
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, blocker.Rollback(context.WithoutCancel(t.Context())))
+	})
+	_, err = blocker.Exec(t.Context(), fmt.Sprintf("LOCK TABLE %s.t IN ACCESS EXCLUSIVE MODE", schema))
+	require.NoError(t, err)
+
+	st := mustParse(t, fmt.Sprintf("ALTER TABLE %s.t ADD COLUMN age int", schema))
+	retry := executor.RetryPolicy{MaxAttempts: 2, InitialBackoff: 10 * time.Millisecond, MaxBackoff: 20 * time.Millisecond}
+	tight := executor.Budget{LockTimeout: 100 * time.Millisecond, StatementTimeout: time.Second}
+	err = executor.ExecuteNativeWithProgress(t.Context(), pool, pt, st, tight, retry, tracker)
+
+	var budgetErr *executor.BudgetError
+	require.ErrorAs(t, err, &budgetErr)
+	assert.Equal(t, executor.CauseLock, budgetErr.Cause)
+	snapshot, err := tracker.Progress(t.Context())
+	require.NoError(t, err)
+	assert.Equal(t, progress.PhaseFailed, snapshot.Phase)
+	assert.Equal(t, retry.MaxAttempts, snapshot.Detail.Attempt,
+		"the tracker must have observed the final bounded attempt")
+	assert.False(t, snapshot.Detail.Active)
+}
+
 // Sub-millisecond budgets are as unbounded as zero ones: they truncate to
 // PostgreSQL's 0ms, which disables the corresponding limit entirely.
 func TestExecuteNativeRejectsUnboundedBudgets(t *testing.T) {
diff --git a/pkg/executor/retry_internal_test.go b/pkg/executor/retry_internal_test.go
index f0a0761..9e0e62c 100644
--- a/pkg/executor/retry_internal_test.go
+++ b/pkg/executor/retry_internal_test.go
@@ -80,3 +80,37 @@ func TestRetryPolicyRejectsUnboundedValues(t *testing.T) {
 func TestRetryPolicyAcceptsSingleAttemptWithoutBackoff(t *testing.T) {
 	require.NoError(t, RetryPolicy{MaxAttempts: 1}.validate())
 }
+
+// The observer sees each attempt number before that attempt runs, so a
+// progress tracker always reports the attempt actually executing.
+func TestExecuteWithLockRetryObservedReportsEachAttempt(t *testing.T) {
+	policy := RetryPolicy{MaxAttempts: 3, InitialBackoff: time.Millisecond, MaxBackoff: time.Millisecond}
+	var observed []int
+	attempts := 0
+	err := executeWithLockRetryObserved(t.Context(), policy, func(context.Context) error {
+		attempts++
+		if attempts < 3 {
+			return &BudgetError{Cause: CauseLock, Budget: time.Millisecond}
+		}
+		return nil
+	}, func(context.Context, time.Duration) error { return nil }, func(attempt int) {
+		require.Equal(t, attempts+1, attempt, "the observer must run before its attempt")
+		observed = append(observed, attempt)
+	})
+	require.NoError(t, err)
+	assert.Equal(t, []int{1, 2, 3}, observed)
+}
+
+// A non-retryable failure still observes its one attempt: the tracker's
+// attempt counter must match what ran, not what succeeded.
+func TestExecuteWithLockRetryObservedReportsFailedOnlyAttempt(t *testing.T) {
+	var observed []int
+	err := executeWithLockRetryObserved(t.Context(), DefaultRetryPolicy(), func(context.Context) error {
+		return &BudgetError{Cause: CauseStatement, Budget: time.Second}
+	}, func(context.Context, time.Duration) error { return errors.New("unexpected sleep") }, func(attempt int) {
+		observed = append(observed, attempt)
+	})
+	var budgetErr *BudgetError
+	require.ErrorAs(t, err, &budgetErr)
+	assert.Equal(t, []int{1}, observed)
+}
diff --git a/pkg/executor/sequence.go b/pkg/executor/sequence.go
index 6e5f59a..29501c9 100644
--- a/pkg/executor/sequence.go
+++ b/pkg/executor/sequence.go
@@ -237,6 +237,17 @@ func RunSequenceWithProgress(ctx context.Context, pool *pgxpool.Pool, pt preflig
 	return runSequence(ctx, pool, pt, steps, b, retry, tracker)
 }
 
+// elapsedSince reports the time since start on the tracker's injected clock
+// when one is present, falling back to the wall clock. Both the step report
+// and the typed budget corroboration must read the same clock that produced
+// start, or an injected test clock would skew the elapsed value.
+func elapsedSince(tracker *progress.Tracker, start time.Time) time.Duration {
+	if tracker != nil {
+		return tracker.Now().Sub(start)
+	}
+	return time.Since(start)
+}
+
 func runSequence(ctx context.Context, pool *pgxpool.Pool, pt preflight.PreflightedTable, steps []string, b SequenceBudget, retry RetryPolicy, tracker *progress.Tracker) (SequenceReport, error) {
 	var rep SequenceReport
 	if err := b.validate(); err != nil {
@@ -293,7 +304,7 @@ func runSequence(ctx context.Context, pool *pgxpool.Pool, pt preflight.Preflight
 				LockTimeout:      b.Validate.LockTimeout,
 				StatementTimeout: b.Validate.Overall,
 			}, retry, tracker)
-			err = corroborateValidateCancel(err, b.Validate, time.Since(start))
+			err = corroborateValidateCancel(err, b.Validate, elapsedSince(tracker, start))
 		case StepBrief:
 			err = executeNative(ctx, pool, pt, step.st, b.Brief, retry, tracker)
 		default:
@@ -304,10 +315,7 @@ func runSequence(ctx context.Context, pool *pgxpool.Pool, pt preflight.Preflight
 		if err != nil {
 			return rep, &SequenceStepError{Step: i + 1, Total: len(admitted), Kind: step.kind, SQL: step.st.SQL(), Err: err}
 		}
-		duration := time.Since(start)
-		if tracker != nil {
-			duration = tracker.Now().Sub(start)
-		}
+		duration := elapsedSince(tracker, start)
 		rep.Steps = append(rep.Steps, StepReport{
 			SQL:      step.st.SQL(),
 			Kind:     step.kind,
diff --git a/pkg/executor/sequence_integration_test.go b/pkg/executor/sequence_integration_test.go
index b821f85..5ab9cc0 100644
--- a/pkg/executor/sequence_integration_test.go
+++ b/pkg/executor/sequence_integration_test.go
@@ -17,6 +17,7 @@ import (
 	"github.com/block/pg-sprite/pkg/dbconn"
 	"github.com/block/pg-sprite/pkg/executor"
 	"github.com/block/pg-sprite/pkg/planner"
+	"github.com/block/pg-sprite/pkg/progress"
 )
 
 // sqlstateCheckViolation is the typed outcome a failed VALIDATE surfaces.
@@ -88,6 +89,65 @@ func TestRunSequenceValidatesCheckConstraintOnline(t *testing.T) {
 	assert.True(t, validated, "the constraint must be validated, not left NOT VALID")
 }
 
+// The tracker must follow the sequence step by step — position, execution
+// class per step kind, and a finished terminal phase — through the public
+// RunSequenceWithProgress entry point.
+func TestRunSequenceWithProgressTracksStepsAndFinishes(t *testing.T) {
+	pool, schema := newPool(t)
+	_, err := pool.Exec(t.Context(), fmt.Sprintf(
+		"CREATE TABLE %s.t (id int PRIMARY KEY, v int); INSERT INTO %s.t SELECT g, g FROM generate_series(1, 100) g",
+		schema, schema))
+	require.NoError(t, err)
+	pt := mustPreflight(t, pool, schema, "t")
+	tracker, err := progress.NewTracker(progress.WallClock{})
+	require.NoError(t, err)
+
+	steps := saferSequence(t, fmt.Sprintf("ALTER TABLE %s.t ADD CONSTRAINT v_positive CHECK (v > 0)", schema))
+	rep, err := executor.RunSequenceWithProgress(t.Context(), pool, pt, steps, runBudget,
+		executor.DefaultRetryPolicy(), tracker)
+	require.NoError(t, err)
+	require.Len(t, rep.Steps, 2)
+
+	snapshot, err := tracker.Progress(t.Context())
+	require.NoError(t, err)
+	assert.Equal(t, progress.PhaseFinished, snapshot.Phase)
+	assert.Equal(t, 2, snapshot.Step, "the tracker must have advanced to the final step")
+	assert.Equal(t, 2, snapshot.TotalSteps)
+	assert.Equal(t, progress.OperationValidate, snapshot.Detail.Operation,
+		"the last step's execution class must be the validation scan")
+	assert.False(t, snapshot.Detail.Active)
+}
+
+// A failing step leaves the tracker in a failed terminal phase still
+// pointing at the step that failed — the observable counterpart of the
+// typed SequenceStepError.
+func TestRunSequenceWithProgressReportsFailedStep(t *testing.T) {
+	pool, schema := newPool(t)
+	_, err := pool.Exec(t.Context(), fmt.Sprintf(
+		"CREATE TABLE %s.t (id int PRIMARY KEY, v int); INSERT INTO %s.t VALUES (1, -1)",
+		schema, schema))
+	require.NoError(t, err)
+	pt := mustPreflight(t, pool, schema, "t")
+	tracker, err := progress.NewTracker(progress.WallClock{})
+	require.NoError(t, err)
+
+	// The violating row makes step 1 (the NOT VALID add) succeed and step 2
+	// (the validation scan) fail.
+	steps := saferSequence(t, fmt.Sprintf("ALTER TABLE %s.t ADD CONSTRAINT v_positive CHECK (v > 0)", schema))
+	_, err = executor.RunSequenceWithProgress(t.Context(), pool, pt, steps, runBudget,
+		executor.DefaultRetryPolicy(), tracker)
+
+	var stepErr *executor.SequenceStepError
+	require.ErrorAs(t, err, &stepErr)
+	require.Equal(t, 2, stepErr.Step)
+	snapshot, err := tracker.Progress(t.Context())
+	require.NoError(t, err)
+	assert.Equal(t, progress.PhaseFailed, snapshot.Phase)
+	assert.Equal(t, stepErr.Step, snapshot.Step, "the tracker must still point at the failed step")
+	assert.Equal(t, 2, snapshot.TotalSteps)
+	assert.False(t, snapshot.Detail.Active)
+}
+
 func TestRunSequenceSetNotNullLeavesNoScaffold(t *testing.T) {
 	pool, schema := newPool(t)
 	_, err := pool.Exec(t.Context(), fmt.Sprintf(
diff --git a/pkg/executor/sequence_internal_test.go b/pkg/executor/sequence_internal_test.go
index 8f33186..12df509 100644
--- a/pkg/executor/sequence_internal_test.go
+++ b/pkg/executor/sequence_internal_test.go
@@ -7,9 +7,12 @@ package executor
 import (
 	"errors"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	"github.com/block/pg-sprite/pkg/progress"
 )
 
 func TestAdmitStepClassifiesShapes(t *testing.T) {
@@ -163,3 +166,29 @@ func TestAdmitSequenceSurfacesParseFailures(t *testing.T) {
 	_, err := admitSequence("s", "t", []string{`ALTER TABLE s.t THIS IS NOT SQL`})
 	require.Error(t, err)
 }
+
+// skewedClock reads a fixed instant far from the wall clock, so any code
+// path that mixes it with time.Now()/time.Since produces a wildly wrong
+// elapsed value instead of a subtly wrong one.
+type skewedClock struct{ now time.Time }
+
+func (c *skewedClock) Now() time.Time { return c.now }
+
+// Elapsed values fed to budget corroboration and step reports must come
+// from the same clock that produced the start instant: a tracker's injected
+// clock when present, the wall clock otherwise.
+func TestElapsedSinceReadsTheClockThatProducedStart(t *testing.T) {
+	clock := &skewedClock{now: time.Unix(1000, 0)}
+	tracker, err := progress.NewTracker(clock)
+	require.NoError(t, err)
+
+	start := tracker.Now()
+	clock.now = clock.now.Add(5 * time.Second)
+	assert.Equal(t, 5*time.Second, elapsedSince(tracker, start),
+		"with a tracker, elapsed must be measured on its injected clock")
+
+	wallStart := time.Now()
+	elapsed := elapsedSince(nil, wallStart)
+	assert.GreaterOrEqual(t, elapsed, time.Duration(0))
+	assert.Less(t, elapsed, time.Minute, "without a tracker, elapsed must be wall-clock time since start")
+}
diff --git a/pkg/executor/withprogress_test.go b/pkg/executor/withprogress_test.go
new file mode 100644
index 0000000..363fb6c
--- /dev/null
+++ b/pkg/executor/withprogress_test.go
@@ -0,0 +1,36 @@
+package executor_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/block/pg-sprite/pkg/executor"
+	"github.com/block/pg-sprite/pkg/preflight"
+	"github.com/block/pg-sprite/pkg/statement"
+)
+
+// The *WithProgress entry points exist for callers that poll; a nil tracker
+// is a caller bug they must refuse with a typed error before anything else
+// runs — never a panic, and never a silent fallback to unobserved execution.
+
+func TestExecuteNativeWithProgressRequiresTracker(t *testing.T) {
+	err := executor.ExecuteNativeWithProgress(t.Context(), nil, preflight.PreflightedTable{},
+		statement.Statement{}, executor.Budget{LockTimeout: time.Second, StatementTimeout: time.Second},
+		executor.DefaultRetryPolicy(), nil)
+	require.Error(t, err)
+}
+
+func TestRunSequenceWithProgressRequiresTracker(t *testing.T) {
+	_, err := executor.RunSequenceWithProgress(t.Context(), nil, preflight.PreflightedTable{},
+		[]string{"ALTER TABLE s.t ADD COLUMN v int"},
+		executor.SequenceBudget{}, executor.DefaultRetryPolicy(), nil)
+	require.Error(t, err)
+}
+
+func TestBuildIndexConcurrentlyWithProgressRequiresTracker(t *testing.T) {
+	_, err := executor.BuildIndexConcurrentlyWithProgress(t.Context(), nil,
+		"CREATE INDEX CONCURRENTLY i ON s.t (c)", executor.ConcurrentBudget{Overall: time.Minute}, nil)
+	require.Error(t, err)
+}
diff --git a/pkg/progress/progress.go b/pkg/progress/progress.go
index e793d91..1cb7740 100644
--- a/pkg/progress/progress.go
+++ b/pkg/progress/progress.go
@@ -86,8 +86,15 @@ type Snapshot struct {
 // Tracker is a concurrency-safe progress source. The caller owns it; it has
 // no goroutines. Progress performs the one read needed for an active index
 // build, making polling lifetime identical to the caller's context.
+//
+// Two locks split the tracker's concerns: mu guards the state fields and is
+// held only for memory access, so the executor's own updates never wait for
+// a database read; pollMu serializes observers, so the reserved session —
+// a single pgx connection that is not safe for concurrent use — only ever
+// carries one progress query at a time.
 type Tracker struct {
 	mu        sync.RWMutex
+	pollMu    sync.Mutex
 	clock     Clock
 	session   dbconn.RowQuerier
 	phase     Phase
@@ -147,9 +154,11 @@ func (t *Tracker) SetConcurrentBuild(session dbconn.RowQuerier, pid uint32) {
 // StopConcurrentBuild waits for an in-flight observation and releases the
 // reserved session back to the executor before its catalog verdict.
 func (t *Tracker) StopConcurrentBuild() {
+	t.pollMu.Lock()
+	defer t.pollMu.Unlock()
 	t.mu.Lock()
 	defer t.mu.Unlock()
-	t.buildPID = 0
+	t.session, t.buildPID = nil, 0
 }
 
 // Finish records a terminal execution outcome.
@@ -166,10 +175,15 @@ func (t *Tracker) Finish(err error) {
 }
 
 // Progress returns a snapshot and, for an active concurrent index build,
-// queries PostgreSQL's progress view by the executor-owned backend PID.
+// queries PostgreSQL's progress view by the executor-owned backend PID. On a
+// query error the snapshot still carries the last-known tracker state. The
+// state lock is released before the query, so concurrent pollers serialize
+// only against each other (and StopConcurrentBuild), never against the
+// executor's own state updates.
 func (t *Tracker) Progress(ctx context.Context) (Snapshot, error) {
+	t.pollMu.Lock()
+	defer t.pollMu.Unlock()
 	t.mu.RLock()
-	defer t.mu.RUnlock()
 	now := t.clock.Now()
 	s := Snapshot{Phase: t.phase, Step: t.step, TotalSteps: t.total, Detail: t.detail}
 	if !t.started.IsZero() {
@@ -177,12 +191,13 @@ func (t *Tracker) Progress(ctx context.Context) (Snapshot, error) {
 		s.StepElapsed = now.Sub(t.stepStart)
 	}
 	session, pid := t.session, t.buildPID
+	t.mu.RUnlock()
 	if pid == 0 || session == nil || s.Phase != PhaseRunning {
 		return s, nil
 	}
 	p, active, err := dbconn.ConcurrentIndexProgress(ctx, session, pid)
 	if err != nil {
-		return Snapshot{}, err
+		return s, err
 	}
 	if !active {
 		s.Detail.Active = false
diff --git a/pkg/progress/progress_test.go b/pkg/progress/progress_test.go
index 1af2faf..855adea 100644
--- a/pkg/progress/progress_test.go
+++ b/pkg/progress/progress_test.go
@@ -1,9 +1,14 @@
 package progress_test
 
 import (
+	"context"
+	"errors"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
+	"github.com/jackc/pgx/v5"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -14,6 +19,33 @@ type fakeClock struct{ now time.Time }
 
 func (c *fakeClock) Now() time.Time { return c.now }
 
+// fakeRow satisfies pgx.Row with a caller-supplied Scan.
+type fakeRow struct{ scan func(dest ...any) error }
+
+func (r fakeRow) Scan(dest ...any) error { return r.scan(dest...) }
+
+// fakeSession satisfies dbconn.RowQuerier with a caller-supplied query,
+// standing in for the executor's reserved verdict session.
+type fakeSession struct {
+	query func(ctx context.Context, sql string, args ...any) pgx.Row
+}
+
+func (s fakeSession) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row {
+	return s.query(ctx, sql, args...)
+}
+
+// runningTrackerWithBuild returns a tracker mid concurrent index build,
+// polling against session.
+func runningTrackerWithBuild(t *testing.T, session fakeSession) *progress.Tracker {
+	t.Helper()
+	tracker, err := progress.NewTracker(&fakeClock{now: time.Unix(100, 0)})
+	require.NoError(t, err)
+	tracker.Start(1, progress.OperationConcurrentIndex)
+	tracker.StartStep(1, progress.OperationConcurrentIndex)
+	tracker.SetConcurrentBuild(session, 4242)
+	return tracker
+}
+
 func TestTrackerReportsSequencePositionAndInjectedElapsed(t *testing.T) {
 	clock := &fakeClock{now: time.Unix(100, 0)}
 	tracker, err := progress.NewTracker(clock)
@@ -73,3 +105,179 @@ func TestNewTrackerRequiresClock(t *testing.T) {
 	_, err := progress.NewTracker(nil)
 	require.Error(t, err)
 }
+
+// The server row's columns must land on the right Work fields — distinct
+// values for every column so a swapped pair cannot pass.
+func TestProgressMergesServerIndexBuildWork(t *testing.T) {
+	session := fakeSession{query: func(context.Context, string, ...any) pgx.Row {
+		return fakeRow{scan: func(dest ...any) error {
+			*(dest[0].(*string)) = "building index"
+			*(dest[1].(*uint64)) = 11 // blocks_done
+			*(dest[2].(*uint64)) = 40 // blocks_total
+			*(dest[3].(*uint64)) = 7  // tuples_done
+			*(dest[4].(*uint64)) = 21 // tuples_total
+			return nil
+		}}
+	}}
+	tracker := runningTrackerWithBuild(t, session)
+
+	s, err := tracker.Progress(t.Context())
+	require.NoError(t, err)
+	assert.True(t, s.Detail.Active)
+	assert.Equal(t, "building index", s.Detail.ServerPhase)
+	require.NotNil(t, s.Detail.Work)
+	assert.Equal(t, uint64(11), s.Detail.Work.BlocksDone)
+	assert.Equal(t, uint64(40), s.Detail.Work.BlocksTotal)
+	assert.Equal(t, uint64(7), s.Detail.Work.TuplesDone)
+	assert.Equal(t, uint64(21), s.Detail.Work.TuplesTotal)
+	assert.Zero(t, s.Detail.Work.RowsCopied, "native progress must not fabricate copy counters")
+}
+
+// A build that has left the progress view is reported inactive, with no
+// stale server detail attached.
+func TestProgressClearsActiveWhenServerRowIsGone(t *testing.T) {
+	session := fakeSession{query: func(context.Context, string, ...any) pgx.Row {
+		return fakeRow{scan: func(...any) error { return pgx.ErrNoRows }}
+	}}
+	tracker := runningTrackerWithBuild(t, session)
+
+	s, err := tracker.Progress(t.Context())
+	require.NoError(t, err)
+	assert.False(t, s.Detail.Active, "a vanished progress row means the build is no longer active")
+	assert.Empty(t, s.Detail.ServerPhase)
+	assert.Nil(t, s.Detail.Work)
+}
+
+// A transient query failure surfaces the error alongside the last-known
+// tracker state, not a zero-valued snapshot.
+func TestProgressReturnsSnapshotAlongsideQueryError(t *testing.T) {
+	queryErr := errors.New("connection severed")
+	session := fakeSession{query: func(context.Context, string, ...any) pgx.Row {
+		return fakeRow{scan: func(...any) error { return queryErr }}
+	}}
+	tracker := runningTrackerWithBuild(t, session)
+
+	s, err := tracker.Progress(t.Context())
+	require.ErrorIs(t, err, queryErr)
+	assert.Equal(t, progress.PhaseRunning, s.Phase, "the snapshot must keep the last-known state on error")
+	assert.Equal(t, 1, s.Step)
+	assert.Equal(t, 1, s.TotalSteps)
+}
+
+// Two concurrent pollers must never drive the reserved session at the same
+// time: a single pgx connection is not safe for concurrent use.
+func TestProgressSerializesConcurrentPollers(t *testing.T) {
+	firstEntered := make(chan struct{})
+	overlap := make(chan struct{})
+	release := make(chan struct{})
+	var entries atomic.Int32
+	session := fakeSession{query: func(context.Context, string, ...any) pgx.Row {
+		return fakeRow{scan: func(...any) error {
+			switch entries.Add(1) {
+			case 1:
+				close(firstEntered)
+			case 2:
+				close(overlap)
+			}
+			<-release
+			return pgx.ErrNoRows
+		}}
+	}}
+	tracker := runningTrackerWithBuild(t, session)
+
+	var pollers sync.WaitGroup
+	for range 2 {
+		pollers.Go(func() {
+			_, err := tracker.Progress(t.Context())
+			assert.NoError(t, err)
+		})
+	}
+	<-firstEntered
+	secondPollerMustStillWait := time.After(100 * time.Millisecond)
+	select {
+	case <-overlap:
+		t.Fatal("two pollers reached the session concurrently")
+	case <-secondPollerMustStillWait:
+	}
+	close(release)
+	pollers.Wait()
+	assert.Equal(t, int32(2), entries.Load(), "both pollers must complete, one after the other")
+}
+
+// StopConcurrentBuild must drain an in-flight observation before returning:
+// the executor reuses the reserved session for its catalog verdict as soon
+// as StopConcurrentBuild returns.
+func TestStopConcurrentBuildDrainsInFlightObservation(t *testing.T) {
+	entered := make(chan struct{})
+	release := make(chan struct{})
+	var queryFinished atomic.Bool
+	session := fakeSession{query: func(context.Context, string, ...any) pgx.Row {
+		return fakeRow{scan: func(...any) error {
+			close(entered)
+			<-release
+			queryFinished.Store(true)
+			return pgx.ErrNoRows
+		}}
+	}}
+	tracker := runningTrackerWithBuild(t, session)
+
+	var workers sync.WaitGroup
+	workers.Go(func() {
+		_, err := tracker.Progress(t.Context())
+		assert.NoError(t, err)
+	})
+	<-entered
+
+	stopReturned := make(chan struct{})
+	workers.Go(func() {
+		tracker.StopConcurrentBuild()
+		assert.True(t, queryFinished.Load(),
+			"StopConcurrentBuild must not return while an observation still holds the session")
+		close(stopReturned)
+	})
+	stopMustStillBlock := time.After(100 * time.Millisecond)
+	select {
+	case <-stopReturned:
+		t.Fatal("StopConcurrentBuild returned while an observation was in flight")
+	case <-stopMustStillBlock:
+	}
+	close(release)
+	workers.Wait()
+}
+
+// The executor's own state updates must never wait behind a slow
+// observation: polling is observability, not a gate on execution.
+func TestStateMutatorsDoNotWaitForInFlightObservation(t *testing.T) {
+	entered := make(chan struct{})
+	release := make(chan struct{})
+	session := fakeSession{query: func(context.Context, string, ...any) pgx.Row {
+		return fakeRow{scan: func(...any) error {
+			close(entered)
+			<-release
+			return pgx.ErrNoRows
+		}}
+	}}
+	tracker := runningTrackerWithBuild(t, session)
+
+	var workers sync.WaitGroup
+	defer workers.Wait()
+	defer close(release)
+	workers.Go(func() {
+		_, err := tracker.Progress(t.Context())
+		assert.NoError(t, err)
+	})
+	<-entered
+
+	mutated := make(chan struct{})
+	workers.Go(func() {
+		tracker.SetAttempt(2)
+		tracker.StartStep(1, progress.OperationConcurrentIndex)
+		close(mutated)
+	})
+	mutatorDeadline := time.After(5 * time.Second)
+	select {
+	case <-mutated:
+	case <-mutatorDeadline:
+		t.Fatal("a state mutator waited behind an in-flight observation")
+	}
+}

From 5cce91237e74bbc774914815083642c8a5fde450 Mon Sep 17 00:00:00 2001
From: Kiran Muddukrishna <kmuddukrishna@squareup.com>
Date: Tue, 18 Aug 2026 09:23:43 +1000
Subject: [PATCH 3/3] Version the progress contract and pin the TCB import
 boundary
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Closes the remaining PR #41 review findings: snapshots carry
format_version with a key-pinning contract test and docs page; terminal
snapshots freeze elapsed at Finish; the tracker fully resets between
runs and drops the build session on step/finish; sequences report an
"admitting" operation before step 1; nil-tracker guards return
ErrInvariantViolation. pkg/progress is reclassified as core in
SAFETY.md — the executors import it — and depguard now mechanically
enforces the recorded core dependency list the docs already claimed.
---
 .golangci.yml                     |  20 +++++
 SAFETY.md                         |   8 +-
 docs/README.md                    |   1 +
 docs/progress-report.md           | 117 +++++++++++++++++++++++++
 docs/tcb-model.md                 |   7 +-
 pkg/executor/native.go            |   2 +-
 pkg/executor/optimistic.go        |   2 +-
 pkg/executor/sequence.go          |   4 +-
 pkg/executor/withprogress_test.go |  12 +--
 pkg/progress/progress.go          |  76 ++++++++++------
 pkg/progress/progress_test.go     | 141 ++++++++++++++++++++++++++++++
 11 files changed, 350 insertions(+), 40 deletions(-)
 create mode 100644 docs/progress-report.md

diff --git a/.golangci.yml b/.golangci.yml
index 026d2ff..0299220 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -22,7 +22,27 @@ linters:
     - containedctx    # no context.Context stored in struct fields
     - sloglint        # structured-logging hygiene: static messages, snake_case keys
     - forbidigo       # no printing to process stdout; output goes to the injected writer
+    - depguard        # pin the TCB import boundary (SAFETY.md core dependency list)
   settings:
+    depguard:
+      rules:
+        # The safety-critical core may import only the recorded core dependency
+        # list (SAFETY.md): stdlib, pgx/v5, the parse boundary, and the other
+        # core packages. Anything else needs a recorded decision there first.
+        tcb:
+          files:
+            - "**/pkg/dbconn/**"
+            - "**/pkg/preflight/**"
+            - "**/pkg/executor/**"
+            - "**/pkg/progress/**"
+            - "!$test"
+          allow:
+            - $gostd
+            - github.com/jackc/pgx/v5
+            - github.com/block/pg-sprite/pkg/dbconn
+            - github.com/block/pg-sprite/pkg/preflight
+            - github.com/block/pg-sprite/pkg/progress
+            - github.com/block/pg-sprite/pkg/statement
     sloglint:
       static-msg: true
       key-naming-case: snake
diff --git a/SAFETY.md b/SAFETY.md
index aa6fe36..3ebeb82 100644
--- a/SAFETY.md
+++ b/SAFETY.md
@@ -31,7 +31,7 @@ The invariant registry (invariant IDs referenced below) lives in
 | `pkg/verdict` — structured outcome contract, rendering, exit codes | ❌ periphery | exists (Phase 1) | — |
 | `pkg/diffplan` — desired schema → routed convergence plan, the declarative front door as a library (the CLI `diff` and embedding orchestrators share it) | ❌ periphery | exists | — |
 | `internal/cli` — CLI, flags, help, prompts | ❌ periphery | `migrate`, `status`, `diff`, `fmt`, `lint`, and `suggest` exist | — |
-| `pkg/progress` — strategy-wide native progress snapshots; copy counters reserved for later | ❌ periphery | native progress exists | — |
+| `pkg/progress` — strategy-wide progress snapshots; the executors' observation seam (core imports it, so its locking discipline is core-critical); copy counters reserved for later | ✅ core | native progress exists | — |
 | orchestrator adapter | ❌ periphery | planned (Phase 11) | OC-* hold *at* the boundary |
 | `internal/testutil` | ❌ test-only | exists | — |
 
@@ -64,8 +64,10 @@ The short version — the full rules live in [docs/tcb-model.md](docs/tcb-model.
   caller's classification; the grammar is load-bearing expertise, not copyable mechanics),
   `pkg/progress` (the executors' progress-observation seam: they write state into a
   caller-owned tracker whose mutators take only a memory lock, and its polling reads ride
-  the reserved verdict session behind a separate poll lock — so a slow or hung observation
-  can never gate the executor's own state updates),
+  the reserved verdict session behind a separate poll lock — the executor's own state
+  updates never wait for a database read, but the verdict handoff *is* observer-gated:
+  `StopConcurrentBuild` deliberately drains an in-flight poll before the executor reclaims
+  the session, a wait bounded by the poller's context and the session's `statement_timeout`),
   stdlib. The future decode path will add `pglogrepl`. Adding one requires a recorded decision (see the rubric in
   [docs/tcb-model.md](docs/tcb-model.md) — copy small things, take pinned dependencies only
   for load-bearing expertise).
diff --git a/docs/README.md b/docs/README.md
index 6a9a726..7fdad27 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -36,6 +36,7 @@ Aurora-only. Why that combination is the product is [vision.md](vision.md); star
 | [limitations.md](limitations.md) | The **current limitations** — schema changes pg-sprite refuses today, why they are unsafe or unsupported, and where an operator must act outside the engine. |
 | [lint-report.md](lint-report.md) | The **lint report contract** — the versioned JSON shape `pg-sprite lint` emits for offline CI gating: finding fields (verbatim SQL, line/column), the codes table, severities and exit behavior, the offline-conservatism rules, and how the contract versions relative to the plan report. |
 | [suggest-report.md](suggest-report.md) | The **suggest report contract** — the versioned JSON shape `pg-sprite suggest` emits for offline advice: the typed caveat vocabulary (what changes about how you must run a safer form, and what a failed step leaves behind), the typed guidance codes for rewrites the planner cannot construct, and the operation → safer form → caveats table (pinned by test). |
+| [progress-report.md](progress-report.md) | The **progress report contract** — the versioned JSON snapshot a caller receives when polling a running change through the `*WithProgress` entry points: phases and operations vocabularies, the terminal-freeze rule, server-observed work counters, and polling semantics (pinned by test). |
 | [engine-role.md](engine-role.md) | The **engine-role provisioning contract** — the tiered minimum access a PostgreSQL user needs to run schema changes against tables it does not own: role membership for owner-gated DDL, schema `CREATE` for index builds and shadow objects, `SET ROLE` for owner-correct shadow creation, replication access for CDC, and the explicit list of powers the engine role must *not* have. Preflight refusals name the missing `GRANT` and point here. |
 | [invalid-index-recovery.md](invalid-index-recovery.md) | The **operator runbook** for the one native-path outcome that needs a human — an invalid index the executor found or may have left. What each typed state licenses: when `DROP INDEX CONCURRENTLY` is proven safe, when the entry may be another actor's healthy in-flight build, and what to check when the executor could prove nothing. |
 | [testing.md](testing.md) | The **test-suite guide** — how to run the suite (unit, per-major, all supported majors, compose database), current coverage, the remaining executor-phase test obligations, and the vanilla-PostgreSQL-matrix vs real-Aurora validation boundary. |
diff --git a/docs/progress-report.md b/docs/progress-report.md
new file mode 100644
index 0000000..0314014
--- /dev/null
+++ b/docs/progress-report.md
@@ -0,0 +1,117 @@
+# The progress report contract
+
+The progress snapshot is the machine-readable observation a caller receives when it polls a
+running schema change through the `*WithProgress` executor entry points. It is the one JSON
+shape an operator or orchestrator consumes to display or act on execution progress. This
+document is the contract: the fields, the closed vocabularies, and the behavior required of
+a consumer. The Go source of truth is `pkg/progress`; `TestSnapshotJSONShape` pins the exact
+keys, including the example at the end of this page.
+
+## Versioning: `format_version`
+
+Every snapshot carries `format_version`. A consumer that does not recognize the version must
+**reject the snapshot** — never guess at field semantics. The version covers more than the
+field shape: the closed vocabularies below (phases, operations) are pinned to it. Adding a
+phase or operation value is a contract change and bumps `format_version`, even if no field
+is added or renamed.
+
+The [plan report](plan-report.md), [lint report](lint-report.md), and
+[suggest report](suggest-report.md) are separate contracts with their own `format_version`;
+all version independently.
+
+## Consumer behavior for unknown values
+
+`phase` and `detail.operation` draw from the closed vocabularies below. A consumer that
+meets a value it does not recognize must treat the execution's state as **unknown** — never
+map it onto a known value and proceed. Progress is observational: an unknown value never
+licenses a consumer to intervene in the change itself.
+
+## Snapshot fields
+
+| Field | Type | Presence | Meaning |
+|---|---|---|---|
+| `format_version` | int | always | Contract version; reject unknown versions. |
+| `phase` | string | always | Overall execution phase (see Phases). |
+| `step` | int | after the first step starts | 1-based position in a multi-step sequence. Absent before execution reaches step 1. |
+| `total_steps` | int | after `Start` | Number of steps in the execution; `1` for single-statement entry points. |
+| `elapsed_ns` | int | always | Nanoseconds since execution started. For a terminal phase, **frozen** at the instant the outcome was recorded — a late poll reports the execution's duration, not the observation's age. |
+| `step_elapsed_ns` | int | always | Nanoseconds since the current step started; frozen the same way at a terminal phase. |
+| `detail` | object | always | The operation currently executing (below). |
+
+## Detail fields
+
+| Field | Type | Presence | Meaning |
+|---|---|---|---|
+| `operation` | string | once execution starts | The current operation's execution class (see Operations). |
+| `server_phase` | string | active concurrent build only | PostgreSQL's own phase string from `pg_stat_progress_create_index`, verbatim. |
+| `active` | bool | always | Whether an operation is executing now. `false` with `phase: "running"` means a concurrent build's progress row has left the server view. |
+| `attempt` | int | bounded retries only | The current attempt number when the executor is inside its bounded retry loop. |
+| `work` | object | server-observed work only | Present exactly when the server published a progress row; then **every** counter below is present, so a fresh build reports honest zeros rather than an empty object. |
+
+### Work counters
+
+`blocks_done` / `blocks_total` and `tuples_done` / `tuples_total` come from
+`pg_stat_progress_create_index` during a concurrent index build. `rows_copied` /
+`rows_total` and `bytes_copied` / `bytes_total` are reserved for copy-and-swap and are `0`
+on every native operation — the engine never fabricates copy counters.
+
+## Phases
+
+| Value | Meaning |
+|---|---|
+| `pending` | Execution has not started. |
+| `running` | Execution is active. |
+| `finished` | Terminal: completed successfully. |
+| `failed` | Terminal: reached a terminal failure. |
+
+A terminal snapshot is immutable: once `finished` or `failed` is observed, every later poll
+returns the identical snapshot, elapsed values included.
+
+## Operations
+
+| Value | Meaning |
+|---|---|
+| `admitting` | A sequence's steps are still being validated; no statement has run yet. |
+| `optimistic` | One bounded direct native attempt. |
+| `brief` | A brief transactional sequence step. |
+| `validate-constraint` | A constraint-validation scan. |
+| `concurrent-index-build` | A concurrent index build (the one operation with server-observed `work`). |
+
+## Polling semantics
+
+The tracker is caller-owned and has no goroutines or timers: polling lifetime is exactly the
+caller's context. A poll during an active concurrent index build performs one read of the
+server's progress view over the executor's reserved session; every other poll is pure
+memory. On a query error the returned snapshot still carries the last-known tracker state —
+`phase` is never empty — with the error returned alongside for the caller to classify.
+
+## Example
+
+A poll during step 2 of a 3-step sequence, mid concurrent index build:
+
+```json
+{
+  "format_version": 1,
+  "phase": "running",
+  "step": 2,
+  "total_steps": 3,
+  "elapsed_ns": 2750000000,
+  "step_elapsed_ns": 750000000,
+  "detail": {
+    "operation": "concurrent-index-build",
+    "server_phase": "building index",
+    "active": true,
+    "attempt": 2,
+    "work": {
+      "rows_copied": 0,
+      "rows_total": 0,
+      "bytes_copied": 0,
+      "bytes_total": 0,
+      "blocks_done": 11,
+      "blocks_total": 40,
+      "tuples_done": 7,
+      "tuples_total": 21
+    }
+  }
+}
+```
diff --git a/docs/tcb-model.md b/docs/tcb-model.md
index a40ff3e..05a3eaa 100644
--- a/docs/tcb-model.md
+++ b/docs/tcb-model.md
@@ -246,9 +246,10 @@ The AI-assistance posture differs per side of the boundary:
    bitcoin-core discipline).
 2. **`cutover`/`swap` API takes domain types only** — the `VerifiedShadow`/`CleanWatermark`/
    `TableLock` types land with their producing packages (Phases 4–7), not retrofitted.
-3. **The `// INV: <id>` convention has landed; `ErrInvariantViolation`** lands with the
+3. **The `// INV: <id>` convention and `ErrInvariantViolation` have landed** with the
    executor phases.
-4. **The enforcement backlog:** [SAFETY.md](../SAFETY.md) + depguard + CODEOWNERS, the
-   property/fuzz suites per rung above, and the optional TLA+ models for cutover and resume.
+4. **The enforcement backlog:** [SAFETY.md](../SAFETY.md), the depguard import-boundary
+   rule (`.golangci.yml`), and CODEOWNERS have landed; still open are the property/fuzz
+   suites per rung above and the optional TLA+ models for cutover and resume.
 5. **The periphery stays free.** None of this doc applies review friction to status text, CLI
    help, or docs — that's the point of having a boundary.
diff --git a/pkg/executor/native.go b/pkg/executor/native.go
index 3889992..357f164 100644
--- a/pkg/executor/native.go
+++ b/pkg/executor/native.go
@@ -274,7 +274,7 @@ func BuildIndexConcurrently(ctx context.Context, pool *pgxpool.Pool, sql string,
 // tracker. The caller may poll tracker concurrently with this blocking call.
 func BuildIndexConcurrentlyWithProgress(ctx context.Context, pool *pgxpool.Pool, sql string, b ConcurrentBudget, tracker *progress.Tracker) (rep IndexBuildReport, err error) {
 	if tracker == nil {
-		return rep, fmt.Errorf("progress tracker is required")
+		return rep, fmt.Errorf("%w: progress tracker is required", ErrInvariantViolation)
 	}
 	tracker.Start(1, progress.OperationConcurrentIndex)
 	tracker.StartStep(1, progress.OperationConcurrentIndex)
diff --git a/pkg/executor/optimistic.go b/pkg/executor/optimistic.go
index bfeff1c..575756a 100644
--- a/pkg/executor/optimistic.go
+++ b/pkg/executor/optimistic.go
@@ -187,7 +187,7 @@ func ExecuteNative(ctx context.Context, pool *pgxpool.Pool, pt preflight.Preflig
 // tracker. The caller may poll tracker concurrently with this blocking call.
 func ExecuteNativeWithProgress(ctx context.Context, pool *pgxpool.Pool, pt preflight.PreflightedTable, st statement.Statement, b Budget, retry RetryPolicy, tracker *progress.Tracker) (err error) {
 	if tracker == nil {
-		return fmt.Errorf("progress tracker is required")
+		return fmt.Errorf("%w: progress tracker is required", ErrInvariantViolation)
 	}
 	tracker.Start(1, progress.OperationOptimistic)
 	tracker.StartStep(1, progress.OperationOptimistic)
diff --git a/pkg/executor/sequence.go b/pkg/executor/sequence.go
index 29501c9..2c967cf 100644
--- a/pkg/executor/sequence.go
+++ b/pkg/executor/sequence.go
@@ -230,9 +230,9 @@ func RunSequence(ctx context.Context, pool *pgxpool.Pool, pt preflight.Preflight
 // current step and its execution class. The caller may poll concurrently.
 func RunSequenceWithProgress(ctx context.Context, pool *pgxpool.Pool, pt preflight.PreflightedTable, steps []string, b SequenceBudget, retry RetryPolicy, tracker *progress.Tracker) (rep SequenceReport, err error) {
 	if tracker == nil {
-		return rep, fmt.Errorf("progress tracker is required")
+		return rep, fmt.Errorf("%w: progress tracker is required", ErrInvariantViolation)
 	}
-	tracker.Start(len(steps), progress.OperationBrief)
+	tracker.Start(len(steps), progress.OperationAdmitting)
 	defer func() { tracker.Finish(err) }()
 	return runSequence(ctx, pool, pt, steps, b, retry, tracker)
 }
diff --git a/pkg/executor/withprogress_test.go b/pkg/executor/withprogress_test.go
index 363fb6c..83b5d91 100644
--- a/pkg/executor/withprogress_test.go
+++ b/pkg/executor/withprogress_test.go
@@ -12,25 +12,27 @@ import (
 )
 
 // The *WithProgress entry points exist for callers that poll; a nil tracker
-// is a caller bug they must refuse with a typed error before anything else
-// runs — never a panic, and never a silent fallback to unobserved execution.
+// is a caller bug they must refuse with ErrInvariantViolation before anything
+// else runs — never a panic, and never a silent fallback to unobserved
+// execution — so OutcomeCode classifies it as a programmer error, not an
+// operational failure to retry.
 
 func TestExecuteNativeWithProgressRequiresTracker(t *testing.T) {
 	err := executor.ExecuteNativeWithProgress(t.Context(), nil, preflight.PreflightedTable{},
 		statement.Statement{}, executor.Budget{LockTimeout: time.Second, StatementTimeout: time.Second},
 		executor.DefaultRetryPolicy(), nil)
-	require.Error(t, err)
+	require.ErrorIs(t, err, executor.ErrInvariantViolation)
 }
 
 func TestRunSequenceWithProgressRequiresTracker(t *testing.T) {
 	_, err := executor.RunSequenceWithProgress(t.Context(), nil, preflight.PreflightedTable{},
 		[]string{"ALTER TABLE s.t ADD COLUMN v int"},
 		executor.SequenceBudget{}, executor.DefaultRetryPolicy(), nil)
-	require.Error(t, err)
+	require.ErrorIs(t, err, executor.ErrInvariantViolation)
 }
 
 func TestBuildIndexConcurrentlyWithProgressRequiresTracker(t *testing.T) {
 	_, err := executor.BuildIndexConcurrentlyWithProgress(t.Context(), nil,
 		"CREATE INDEX CONCURRENTLY i ON s.t (c)", executor.ConcurrentBudget{Overall: time.Minute}, nil)
-	require.Error(t, err)
+	require.ErrorIs(t, err, executor.ErrInvariantViolation)
 }
diff --git a/pkg/progress/progress.go b/pkg/progress/progress.go
index 1cb7740..2f88189 100644
--- a/pkg/progress/progress.go
+++ b/pkg/progress/progress.go
@@ -37,10 +37,19 @@ const (
 	PhaseFailed Phase = "failed"
 )
 
+// FormatVersion identifies the snapshot contract. A consumer must reject a
+// snapshot whose format_version it does not recognize rather than guess at
+// field semantics. Adding a phase or operation value is a contract change
+// and bumps this version, even when no field is added or renamed.
+const FormatVersion = 1
+
 // Operation is the current operation's execution class.
 type Operation string
 
 const (
+	// OperationAdmitting is the pre-execution window in which a sequence's
+	// steps are still being validated; no statement has run yet.
+	OperationAdmitting Operation = "admitting"
 	// OperationOptimistic is one bounded direct native attempt.
 	OperationOptimistic Operation = "optimistic"
 	// OperationBrief is a brief transactional sequence step.
@@ -51,17 +60,20 @@ const (
 	OperationConcurrentIndex Operation = "concurrent-index-build"
 )
 
-// Work reports optional server-observed work. Rows and bytes are reserved for
-// copy-and-swap; native operations do not fabricate them.
+// Work reports server-observed work. It is present only when the server
+// published a progress row, and then every counter marshals explicitly — a
+// fresh build reports honest zeros, never an empty object a consumer must
+// guess at. Rows and bytes are reserved for copy-and-swap; native operations
+// do not fabricate them.
 type Work struct {
-	RowsCopied  uint64 `json:"rows_copied,omitempty"`
-	RowsTotal   uint64 `json:"rows_total,omitempty"`
-	BytesCopied uint64 `json:"bytes_copied,omitempty"`
-	BytesTotal  uint64 `json:"bytes_total,omitempty"`
-	BlocksDone  uint64 `json:"blocks_done,omitempty"`
-	BlocksTotal uint64 `json:"blocks_total,omitempty"`
-	TuplesDone  uint64 `json:"tuples_done,omitempty"`
-	TuplesTotal uint64 `json:"tuples_total,omitempty"`
+	RowsCopied  uint64 `json:"rows_copied"`
+	RowsTotal   uint64 `json:"rows_total"`
+	BytesCopied uint64 `json:"bytes_copied"`
+	BytesTotal  uint64 `json:"bytes_total"`
+	BlocksDone  uint64 `json:"blocks_done"`
+	BlocksTotal uint64 `json:"blocks_total"`
+	TuplesDone  uint64 `json:"tuples_done"`
+	TuplesTotal uint64 `json:"tuples_total"`
 }
 
 // Detail describes the operation currently executing.
@@ -73,14 +85,17 @@ type Detail struct {
 	Work        *Work     `json:"work,omitempty"`
 }
 
-// Snapshot is one immutable progress observation.
+// Snapshot is one immutable progress observation. For a terminal phase the
+// elapsed values are frozen at the instant Finish recorded, so a late poll
+// reports the execution's duration, not the observation's age.
 type Snapshot struct {
-	Phase       Phase         `json:"phase"`
-	Step        int           `json:"step,omitempty"`
-	TotalSteps  int           `json:"total_steps,omitempty"`
-	Elapsed     time.Duration `json:"elapsed_ns"`
-	StepElapsed time.Duration `json:"step_elapsed_ns,omitempty"`
-	Detail      Detail        `json:"detail"`
+	FormatVersion int           `json:"format_version"`
+	Phase         Phase         `json:"phase"`
+	Step          int           `json:"step,omitempty"`
+	TotalSteps    int           `json:"total_steps,omitempty"`
+	Elapsed       time.Duration `json:"elapsed_ns"`
+	StepElapsed   time.Duration `json:"step_elapsed_ns"`
+	Detail        Detail        `json:"detail"`
 }
 
 // Tracker is a concurrency-safe progress source. The caller owns it; it has
@@ -100,6 +115,7 @@ type Tracker struct {
 	phase     Phase
 	started   time.Time
 	stepStart time.Time
+	ended     time.Time
 	step      int
 	total     int
 	detail    Detail
@@ -117,22 +133,26 @@ func NewTracker(clock Clock) (*Tracker, error) {
 // Now returns the tracker's injected time for executor duration accounting.
 func (t *Tracker) Now() time.Time { return t.clock.Now() }
 
-// Start records the beginning of an execution.
+// Start records the beginning of an execution. It resets all per-execution
+// state, so a reused tracker never leaks a prior run's step, terminal time,
+// or session into the new run's snapshots.
 func (t *Tracker) Start(total int, operation Operation) {
 	now := t.clock.Now()
 	t.mu.Lock()
 	defer t.mu.Unlock()
-	t.phase, t.started, t.stepStart = PhaseRunning, now, now
-	t.total, t.detail = total, Detail{Operation: operation, Active: true}
+	t.phase, t.started, t.stepStart, t.ended = PhaseRunning, now, now, time.Time{}
+	t.step, t.total, t.detail = 0, total, Detail{Operation: operation, Active: true}
+	t.session, t.buildPID = nil, 0
 }
 
-// StartStep advances a sequence to a 1-based step.
+// StartStep advances a sequence to a 1-based step and drops any build
+// session from a prior step, so a later step can never poll a stale build.
 func (t *Tracker) StartStep(step int, operation Operation) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 	t.step, t.stepStart = step, t.clock.Now()
 	t.detail = Detail{Operation: operation, Active: true}
-	t.buildPID = 0
+	t.session, t.buildPID = nil, 0
 }
 
 // SetAttempt records the current bounded retry attempt.
@@ -161,8 +181,10 @@ func (t *Tracker) StopConcurrentBuild() {
 	t.session, t.buildPID = nil, 0
 }
 
-// Finish records a terminal execution outcome.
+// Finish records a terminal execution outcome and the instant it happened;
+// elapsed values in later snapshots freeze at that instant.
 func (t *Tracker) Finish(err error) {
+	now := t.clock.Now()
 	t.mu.Lock()
 	defer t.mu.Unlock()
 	if err == nil {
@@ -170,8 +192,9 @@ func (t *Tracker) Finish(err error) {
 	} else {
 		t.phase = PhaseFailed
 	}
+	t.ended = now
 	t.detail.Active = false
-	t.buildPID = 0
+	t.session, t.buildPID = nil, 0
 }
 
 // Progress returns a snapshot and, for an active concurrent index build,
@@ -185,7 +208,10 @@ func (t *Tracker) Progress(ctx context.Context) (Snapshot, error) {
 	defer t.pollMu.Unlock()
 	t.mu.RLock()
 	now := t.clock.Now()
-	s := Snapshot{Phase: t.phase, Step: t.step, TotalSteps: t.total, Detail: t.detail}
+	if !t.ended.IsZero() {
+		now = t.ended
+	}
+	s := Snapshot{FormatVersion: FormatVersion, Phase: t.phase, Step: t.step, TotalSteps: t.total, Detail: t.detail}
 	if !t.started.IsZero() {
 		s.Elapsed = now.Sub(t.started)
 		s.StepElapsed = now.Sub(t.stepStart)
diff --git a/pkg/progress/progress_test.go b/pkg/progress/progress_test.go
index 855adea..fc7238e 100644
--- a/pkg/progress/progress_test.go
+++ b/pkg/progress/progress_test.go
@@ -2,6 +2,7 @@ package progress_test
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"sync"
 	"sync/atomic"
@@ -106,6 +107,146 @@ func TestNewTrackerRequiresClock(t *testing.T) {
 	require.Error(t, err)
 }
 
+// A terminal snapshot is terminal: elapsed values freeze at the instant
+// Finish recorded and do not grow with the clock, for both outcomes.
+func TestTerminalSnapshotFreezesElapsed(t *testing.T) {
+	cases := []struct {
+		name    string
+		outcome error
+		phase   progress.Phase
+	}{
+		{name: "finished", outcome: nil, phase: progress.PhaseFinished},
+		{name: "failed", outcome: errors.New("build failed"), phase: progress.PhaseFailed},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			clock := &fakeClock{now: time.Unix(100, 0)}
+			tracker, err := progress.NewTracker(clock)
+			require.NoError(t, err)
+			tracker.Start(1, progress.OperationOptimistic)
+			tracker.StartStep(1, progress.OperationOptimistic)
+			clock.now = clock.now.Add(3 * time.Second)
+			tracker.Finish(tc.outcome)
+
+			clock.now = clock.now.Add(time.Hour)
+			snapshot, err := tracker.Progress(t.Context())
+			require.NoError(t, err)
+			assert.Equal(t, tc.phase, snapshot.Phase)
+			assert.Equal(t, 3*time.Second, snapshot.Elapsed, "elapsed must freeze at Finish")
+			assert.Equal(t, 3*time.Second, snapshot.StepElapsed, "step elapsed must freeze at Finish")
+
+			clock.now = clock.now.Add(time.Hour)
+			again, err := tracker.Progress(t.Context())
+			require.NoError(t, err)
+			assert.Equal(t, snapshot, again, "a terminal snapshot must not change between polls")
+		})
+	}
+}
+
+// Start resets everything a prior run left behind: a reused tracker must
+// never report the previous run's step, terminal instant, or build session.
+func TestStartResetsPriorRunState(t *testing.T) {
+	clock := &fakeClock{now: time.Unix(100, 0)}
+	tracker, err := progress.NewTracker(clock)
+	require.NoError(t, err)
+	tracker.Start(3, progress.OperationBrief)
+	tracker.StartStep(2, progress.OperationValidate)
+	tracker.SetConcurrentBuild(fakeSession{query: func(context.Context, string, ...any) pgx.Row {
+		return fakeRow{scan: func(...any) error {
+			t.Fatal("a new run must not poll the prior run's session")
+			return nil
+		}}
+	}}, 4242)
+	tracker.Finish(errors.New("first run failed"))
+
+	clock.now = clock.now.Add(time.Minute)
+	tracker.Start(1, progress.OperationConcurrentIndex)
+	clock.now = clock.now.Add(time.Second)
+
+	snapshot, err := tracker.Progress(t.Context())
+	require.NoError(t, err)
+	assert.Equal(t, progress.PhaseRunning, snapshot.Phase)
+	assert.Zero(t, snapshot.Step, "the prior run's step must not leak into the new run")
+	assert.Equal(t, 1, snapshot.TotalSteps)
+	assert.Equal(t, time.Second, snapshot.Elapsed, "elapsed must restart, not resume from the prior terminal instant")
+	assert.Equal(t, progress.OperationConcurrentIndex, snapshot.Detail.Operation)
+}
+
+// The JSON shape is the adapter-facing contract: exact keys, exact
+// omissions, driven through a real poll so the test pins what a consumer
+// actually receives. A consumer pins format_version 1 against this test.
+func TestSnapshotJSONShape(t *testing.T) {
+	session := fakeSession{query: func(context.Context, string, ...any) pgx.Row {
+		return fakeRow{scan: func(dest ...any) error {
+			*(dest[0].(*string)) = "building index"
+			*(dest[1].(*uint64)) = 11 // blocks_done
+			*(dest[2].(*uint64)) = 40 // blocks_total
+			*(dest[3].(*uint64)) = 7  // tuples_done
+			*(dest[4].(*uint64)) = 21 // tuples_total
+			return nil
+		}}
+	}}
+	clock := &fakeClock{now: time.Unix(100, 0)}
+	tracker, err := progress.NewTracker(clock)
+	require.NoError(t, err)
+	tracker.Start(3, progress.OperationAdmitting)
+	clock.now = clock.now.Add(2 * time.Second)
+	tracker.StartStep(2, progress.OperationConcurrentIndex)
+	tracker.SetAttempt(2)
+	tracker.SetConcurrentBuild(session, 4242)
+	clock.now = clock.now.Add(750 * time.Millisecond)
+
+	snapshot, err := tracker.Progress(t.Context())
+	require.NoError(t, err)
+	raw, err := json.Marshal(snapshot)
+	require.NoError(t, err)
+	assert.JSONEq(t, `{
+		"format_version": 1,
+		"phase": "running",
+		"step": 2,
+		"total_steps": 3,
+		"elapsed_ns": 2750000000,
+		"step_elapsed_ns": 750000000,
+		"detail": {
+			"operation": "concurrent-index-build",
+			"server_phase": "building index",
+			"active": true,
+			"attempt": 2,
+			"work": {
+				"rows_copied": 0,
+				"rows_total": 0,
+				"bytes_copied": 0,
+				"bytes_total": 0,
+				"blocks_done": 11,
+				"blocks_total": 40,
+				"tuples_done": 7,
+				"tuples_total": 21
+			}
+		}
+	}`, string(raw))
+}
+
+// Optional fields are omitted, not emitted as zero values — but the always-on
+// keys (format_version, phase, both elapsed counters, active) are present
+// even on an idle tracker, so a consumer never guesses whether zero means
+// "unset" or "omitted".
+func TestSnapshotJSONOmitsUnsetOptionalFields(t *testing.T) {
+	tracker, err := progress.NewTracker(&fakeClock{now: time.Unix(100, 0)})
+	require.NoError(t, err)
+
+	snapshot, err := tracker.Progress(t.Context())
+	require.NoError(t, err)
+	raw, err := json.Marshal(snapshot)
+	require.NoError(t, err)
+	assert.JSONEq(t, `{
+		"format_version": 1,
+		"phase": "pending",
+		"elapsed_ns": 0,
+		"step_elapsed_ns": 0,
+		"detail": {"active": false}
+	}`, string(raw))
+}
+
 // The server row's columns must land on the right Work fields — distinct
 // values for every column so a swapped pair cannot pass.
 func TestProgressMergesServerIndexBuildWork(t *testing.T) {