diff --git a/pkg/engine/postgres/apply.go b/pkg/engine/postgres/apply.go
index d3201a982..646c27e55 100644
--- a/pkg/engine/postgres/apply.go
+++ b/pkg/engine/postgres/apply.go
@@ -16,6 +16,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"
 	pgstatement "github.com/block/pg-sprite/pkg/statement"
 
 	"github.com/block/schemabot/pkg/ddl"
@@ -39,6 +40,31 @@ const (
 	// catalog verdict — so it only fires on genuine hangs.
 	optimisticApplyCeiling = 5 * time.Minute
 
+	// executorProgressReadTimeout bounds one read of the executor's tracker.
+	// For an active concurrent index build that read is a single-row query
+	// on the session the executor reserved for the build's failure verdict,
+	// and the tracker serializes it against the executor's own end-of-build
+	// fence — so a poll that hangs on a half-open socket would hold up the
+	// apply's verdict and its terminal publish. A deadline of the engine's
+	// own keeps that coupling bounded regardless of what the caller's
+	// context does.
+	//
+	// The bound sits above the statement_timeout the apply pool runs its
+	// sessions under (spritePoolConfig leaves it at pg-sprite's default) so
+	// that on a live connection the server ends a slow read first: a
+	// statement_timeout cancels the query and hands the session back
+	// intact, whereas a client deadline expiring mid-query closes the
+	// connection — and that connection is the one the build's failure
+	// verdict needs. The engine's deadline is therefore the bound of last
+	// resort, for the socket the server's cancellation can no longer reach.
+	executorProgressReadTimeout = dbconn.DefaultStatementTimeout + executorProgressReadHeadroom
+
+	// executorProgressReadHeadroom is how far the read deadline sits past
+	// the session's statement_timeout: long enough that the server's
+	// cancellation error reaches the client before the client gives up on
+	// the socket.
+	executorProgressReadHeadroom = 5 * time.Second
+
 	// concurrentIndexBudget bounds one CREATE INDEX CONCURRENTLY build.
 	// Concurrent builds get their own budget instead of the per-statement
 	// limit: their snapshot waits are lock waits by implementation, so the
@@ -102,16 +128,25 @@ func (e *Engine) Apply(ctx context.Context, req *engine.ApplyRequest) (*engine.A
 		}
 	}
 
-	started := time.Now()
-	key := progressIdentity(req.ResumeState)
-	e.claimProgress(key, progressResult(engine.StateRunning, "preflight", started, change, ""))
-	bgCtx := context.WithoutCancel(ctx)
-	conn := targetConn{dsn: req.Credentials.DSN, caCertPath: caPath}
+	// One tracker per apply: pg-sprite's executor records each step it
+	// starts on it, and Progress reads the position back, so the poller
+	// sees the step in flight rather than only the accept and terminal
+	// states the engine itself publishes.
+	tracker, err := progress.NewTracker(progress.WallClock{})
+	if err != nil {
+		return nil, fmt.Errorf("apply PostgreSQL database %q: %w", req.Database, err)
+	}
+
 	logger := req.Logger
 	if logger == nil {
 		logger = slog.Default()
 	}
-	e.wg.Go(func() { e.runOptimisticApply(bgCtx, conn, change, key, started, logger) })
+	started := time.Now()
+	key := progressIdentity(req.ResumeState)
+	e.claimProgress(key, progressResult(engine.StateRunning, "preflight", started, change, ""), tracker, logger)
+	bgCtx := context.WithoutCancel(ctx)
+	conn := targetConn{dsn: req.Credentials.DSN, caCertPath: caPath}
+	e.wg.Go(func() { e.runOptimisticApply(bgCtx, conn, change, key, started, logger, tracker) })
 
 	return &engine.ApplyResult{
 		Accepted:    true,
@@ -165,14 +200,33 @@ func postgresCreateSetStatements(script string) ([]string, error) {
 	return statements, nil
 }
 
-func (e *Engine) runOptimisticApply(ctx context.Context, conn targetConn, change nativeApply, key string, started time.Time, logger *slog.Logger) {
+func (e *Engine) runOptimisticApply(ctx context.Context, conn targetConn, change nativeApply, key string, started time.Time, logger *slog.Logger, tracker *progress.Tracker) {
 	// The context arrives detached from the caller (an accepted apply must
 	// survive the request), so boundedness comes from the ceiling instead.
 	ctx, cancel := context.WithTimeout(ctx, optimisticApplyCeiling)
 	defer cancel()
-	err := executeOptimistic(ctx, conn, change, e.tableSizeLimit)
+	// Every terminal result carries the executor's final position: the
+	// executor finishes its tracker before it returns, so the read is
+	// memory-only and a failed create reports the step that failed, not the
+	// sequence's first step. The error branch is an invariant guard, not an
+	// expected outcome — a tracker still reporting a live build here means
+	// the executor returned without finishing it, and the terminal result
+	// then carries the tracker's last-known position instead of a fresh
+	// read the reserved session can no longer answer.
+	publish := func(result *engine.ProgressResult) {
+		if err := executorProgressMetadata(ctx, tracker, result.Metadata); err != nil {
+			logger.Warn("PostgreSQL apply terminal progress reports the last-known executor position",
+				"namespace", change.namespace, "table", change.table, "task_id", key, "error", err)
+		}
+		e.publishProgress(key, result, logger)
+	}
+	execute := e.execute
+	if execute == nil {
+		execute = executeOptimistic
+	}
+	err := execute(ctx, conn, change, e.tableSizeLimit, tracker)
 	if err == nil {
-		e.publishProgress(key, progressResult(engine.StateCompleted, "completed", started, change, ""), logger)
+		publish(progressResult(engine.StateCompleted, "completed", started, change, ""))
 		return
 	}
 
@@ -198,7 +252,7 @@ func (e *Engine) runOptimisticApply(ctx context.Context, conn targetConn, change
 		result := progressResult(engine.StateFailed, "failed", started, change,
 			invalidIndexDetail(invalidErr))
 		result.Retryable = true
-		e.publishProgress(key, result, logger)
+		publish(result)
 		return
 	}
 
@@ -210,7 +264,7 @@ func (e *Engine) runOptimisticApply(ctx context.Context, conn targetConn, change
 		// because it is published to GitHub.
 		logger.Warn("PostgreSQL schema change refused",
 			"namespace", change.namespace, "table", change.table, "reason", r.reason, "error", err)
-		e.publishProgress(key, progressResult(engine.StateFailed, "refused", started, change, r.detail), logger)
+		publish(progressResult(engine.StateFailed, "refused", started, change, r.detail))
 		return
 	}
 
@@ -227,7 +281,7 @@ func (e *Engine) runOptimisticApply(ctx context.Context, conn targetConn, change
 	}
 	result := progressResult(engine.StateFailed, "failed", started, change, failure.detail)
 	result.Retryable = failure.retryable
-	e.publishProgress(key, result, logger)
+	publish(result)
 }
 
 // applyFailure is the drive-facing disposition of an operational apply
@@ -595,7 +649,10 @@ func tableNotFoundRefusal(table string) *refusal {
 	}
 }
 
-func executeOptimistic(ctx context.Context, conn targetConn, change nativeApply, tableSizeLimit int64) error {
+// executeOptimistic runs the planned change through pg-sprite's executors,
+// each feeding the tracker so a concurrent Progress poll reads the step and
+// statement in flight.
+func executeOptimistic(ctx context.Context, conn targetConn, change nativeApply, tableSizeLimit int64, tracker *progress.Tracker) error {
 	poolCfg, err := spritePoolConfig(conn.dsn, conn.caCertPath)
 	if err != nil {
 		return fmt.Errorf("prepare pg-sprite apply pool for table %q: %w", change.table, err)
@@ -618,7 +675,7 @@ func executeOptimistic(ctx context.Context, conn targetConn, change nativeApply,
 		// The off-ladder create tier has its own preflight sequence: the
 		// ladder checks below state facts about an existing table, and a
 		// greenfield target has none.
-		return executeCreate(ctx, pool, change, statements)
+		return executeCreate(ctx, pool, change, statements, tracker)
 	}
 	if len(statements) != 1 {
 		return fmt.Errorf("execute PostgreSQL table %q: privilege tier %s requires exactly one statement, got %d", change.table, tier, len(statements))
@@ -654,8 +711,8 @@ func executeOptimistic(ctx context.Context, conn targetConn, change nativeApply,
 		// dedicated index-build executor runs it under the CONCURRENTLY
 		// budget policy and returns a catalog-verified verdict — including
 		// the invalid-index recovery a failed build needs.
-		if _, err := executor.BuildIndexConcurrently(ctx, pool, change.sql,
-			executor.ConcurrentBudget{Overall: concurrentIndexBudget}); err != nil {
+		if _, err := executor.BuildIndexConcurrentlyWithProgress(ctx, pool, change.sql,
+			executor.ConcurrentBudget{Overall: concurrentIndexBudget}, tracker); err != nil {
 			return fmt.Errorf("build PostgreSQL index concurrently on table %q: %w", change.table, err)
 		}
 		return nil
@@ -677,9 +734,9 @@ func executeOptimistic(ctx context.Context, conn targetConn, change nativeApply,
 	// where greenfieldCreateSet keeps the blocking form on purpose — a
 	// table born in the run has no readers, and CONCURRENTLY cannot run
 	// inside its create sequence.
-	if err := executor.ExecuteNative(ctx, pool, table, statement, executor.Budget{
+	if err := executor.ExecuteNativeWithProgress(ctx, pool, table, statement, executor.Budget{
 		LockTimeout: optimisticLockTimeout, StatementTimeout: optimisticStatementLimit,
-	}, executor.DefaultRetryPolicy()); err != nil {
+	}, executor.DefaultRetryPolicy(), tracker); err != nil {
 		return fmt.Errorf("execute native-safe PostgreSQL statement on table %q: %w", change.table, err)
 	}
 	return nil
@@ -692,7 +749,7 @@ func executeOptimistic(ctx context.Context, conn targetConn, change nativeApply,
 // nothing about apply time. The table size gate deliberately does not run:
 // it bounds rewrites of existing data, and a table that does not exist yet
 // has none.
-func executeCreate(ctx context.Context, pool *pgxpool.Pool, change nativeApply, statements []string) error {
+func executeCreate(ctx context.Context, pool *pgxpool.Pool, change nativeApply, statements []string, tracker *progress.Tracker) error {
 	// The planned statements arrive schema-qualified; the desired-schema
 	// contract wants the unqualified form and the executor pins the schema
 	// from the absence proof instead.
@@ -716,9 +773,9 @@ func executeCreate(ctx context.Context, pool *pgxpool.Pool, change nativeApply,
 	if err != nil {
 		return fmt.Errorf("verify PostgreSQL table %q is absent: %w", change.table, err)
 	}
-	if _, err := executor.ExecuteCreate(ctx, pool, absent, role, desired, executor.Budget{
+	if _, err := executor.ExecuteCreateWithProgress(ctx, pool, absent, role, desired, executor.Budget{
 		LockTimeout: optimisticLockTimeout, StatementTimeout: optimisticStatementLimit,
-	}, executor.DefaultRetryPolicy()); err != nil {
+	}, executor.DefaultRetryPolicy(), tracker); err != nil {
 		return fmt.Errorf("execute PostgreSQL CREATE TABLE %q: %w", change.table, err)
 	}
 	return nil
@@ -731,31 +788,100 @@ func executeCreate(ctx context.Context, pool *pgxpool.Pool, change nativeApply,
 // of a target, and answering with whichever apply wrote last would report
 // another schema change's state, including a terminal one, for work that is
 // still in flight. A caller asking about an apply the engine is not tracking
-// gets the idle sentinel. Rich server progress is intentionally absent until
-// the PostgreSQL executor exposes it.
-func (e *Engine) Progress(_ context.Context, req *engine.ProgressRequest) (*engine.ProgressResult, error) {
+// gets the idle sentinel.
+//
+// While the apply runs, the step position and statement come from the
+// pg-sprite tracker its executor feeds; the engine's own record only moves at
+// accept and at the terminal outcome. The tracker read happens outside the
+// engine lock: for an active concurrent index build it queries the server's
+// progress view, and a poll must never hold up Apply or publishProgress on
+// a database round trip. The coupling runs the other way too: the tracker
+// serializes that query against the executor's own end-of-build fence, so
+// a read still on the wire delays the apply's failure verdict and the
+// terminal publish behind it. executorProgressMetadata bounds every read
+// with the engine's own deadline so that delay is never open-ended.
+func (e *Engine) Progress(ctx context.Context, req *engine.ProgressRequest) (*engine.ProgressResult, error) {
 	var key string
 	if req != nil {
 		key = progressIdentity(req.ResumeState)
 	}
 	e.mu.Lock()
-	defer e.mu.Unlock()
 	tracked := e.progress[key]
 	if tracked == nil {
+		e.mu.Unlock()
 		// The exact idle message is a cross-engine contract: stale-task
 		// recovery compares against it verbatim to auto-resolve work
 		// abandoned by a crashed server.
 		return &engine.ProgressResult{State: engine.StatePending, Message: "No active schema change"}, nil
 	}
-	result := *tracked
-	result.Metadata = cloneMetadata(tracked.Metadata)
-	result.Tables = cloneTables(tracked.Tables)
-	if len(result.Tables) > 0 && result.Tables[0].StartedAt != nil && !result.State.IsTerminal() {
+	result := *tracked.result
+	result.Metadata = cloneMetadata(tracked.result.Metadata)
+	result.Tables = cloneTables(tracked.result.Tables)
+	tracker, logger := tracked.tracker, tracked.logger
+	e.mu.Unlock()
+
+	if result.State.IsTerminal() {
+		// A terminal result already carries the executor's final position,
+		// folded in when it was published.
+		return &result, nil
+	}
+	if len(result.Tables) > 0 && result.Tables[0].StartedAt != nil {
 		result.Metadata["elapsed"] = time.Since(*result.Tables[0].StartedAt).Round(time.Millisecond).String()
 	}
+	if err := executorProgressMetadata(ctx, tracker, result.Metadata); err != nil {
+		// The poll still answers with the last-known position: a progress
+		// view the engine cannot read this instant is not a reason to tell
+		// the driver its apply is unobservable.
+		logger.Warn("PostgreSQL apply progress reports the last-known executor position",
+			"task_id", key, "error", err)
+	}
 	return &result, nil
 }
 
+// executorProgressMetadata folds the tracker's current position into the
+// published metadata: the 1-based step the executor is running, the sequence
+// length it announced, and the statement text of that step. Each key is
+// written only once the executor has reported it, so before execution starts
+// the metadata keeps progressResult's pre-execution position. The statement
+// passes through sanitizeStatementText because the metadata is destined for
+// operator-facing single-line rendering and is stored at a bounded width,
+// and the value must satisfy both the moment one starts reading it.
+//
+// For an active concurrent index build the tracker queries the session the
+// executor reserved for the build's failure verdict. The read runs on its
+// own bounded context, detached from the caller's cancellation: a poller or
+// drive context cancelled mid-query would otherwise tear down that session
+// and leave the build's verdict indeterminate. The deadline starts before
+// the tracker takes its observer lock, so the wait behind another observer's
+// read counts against the same budget as the read itself — the poller is not
+// always the apply's own driver, since another apply's conflict probe reads
+// this tracker too, and a read parked behind one must not hold the
+// executor's fence for longer than the engine allows any single read. A
+// tracker read fails only when the server's progress view cannot be queried;
+// the snapshot still carries the last-known position, which is written
+// before the error is returned for the caller to log.
+func executorProgressMetadata(ctx context.Context, tracker *progress.Tracker, metadata map[string]string) error {
+	if tracker == nil {
+		return nil
+	}
+	readCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), executorProgressReadTimeout)
+	defer cancel()
+	snapshot, err := tracker.Progress(readCtx)
+	if snapshot.TotalSteps > 0 {
+		metadata["steps_total"] = strconv.Itoa(snapshot.TotalSteps)
+	}
+	if snapshot.Step > 0 {
+		metadata["step"] = strconv.Itoa(snapshot.Step)
+	}
+	if statement := sanitizeStatementText(snapshot.Detail.Statement); statement != "" {
+		metadata["statement"] = statement
+	}
+	if err != nil {
+		return fmt.Errorf("read pg-sprite executor progress: %w", err)
+	}
+	return nil
+}
+
 // progressIdentity extracts the apply identity that keys engine progress.
 // The drive layer stamps the task identifier into
 // ResumeState.MigrationContext on both Apply and Progress requests, so the
@@ -781,11 +907,13 @@ func progressResult(state engine.State, phase string, started time.Time, change
 		ErrorMessage: detail,
 		Metadata: map[string]string{
 			"phase": phase, "elapsed": time.Since(started).Round(time.Millisecond).String(),
-			// Per-step position is deliberately not tracked yet: the apply
-			// publishes progress only at accept and at the terminal outcome,
-			// and nothing observes the executor's step transitions in
-			// between, so the position stays at the sequence's first step
-			// while the total reports the real create-set length.
+			// The position the record carries before the executor has
+			// reported one: the first step of the planned sequence, with the
+			// total taken from the plan. executorProgressMetadata replaces
+			// each key as the tracker reports it — the total once the
+			// executor announces its sequence, the step once it starts one —
+			// so through the executor's admission checks the record still
+			// shows the planned first step.
 			"step": "1", "steps_total": strconv.Itoa(steps),
 		},
 		Tables: []engine.TableProgress{{
@@ -813,18 +941,18 @@ func progressResult(state engine.State, phase string, started time.Time, change
 // Entries for applies that are still running are never retired, so an
 // in-flight change always answers for itself no matter how many siblings the
 // engine accepts.
-func (e *Engine) claimProgress(key string, result *engine.ProgressResult) {
+func (e *Engine) claimProgress(key string, result *engine.ProgressResult, tracker *progress.Tracker, logger *slog.Logger) {
 	e.mu.Lock()
 	defer e.mu.Unlock()
 	if e.progress == nil {
-		e.progress = make(map[string]*engine.ProgressResult)
+		e.progress = make(map[string]*trackedApply)
 	}
-	for tracked, progress := range e.progress {
-		if tracked != key && progress.State.IsTerminal() {
-			delete(e.progress, tracked)
+	for id, tracked := range e.progress {
+		if id != key && tracked.result.State.IsTerminal() {
+			delete(e.progress, id)
 		}
 	}
-	e.progress[key] = result
+	e.progress[key] = &trackedApply{result: result, tracker: tracker, logger: logger}
 }
 
 // publishProgress stores a background apply's progress unless the engine has
@@ -835,12 +963,13 @@ func (e *Engine) claimProgress(key string, result *engine.ProgressResult) {
 func (e *Engine) publishProgress(key string, result *engine.ProgressResult, logger *slog.Logger) {
 	e.mu.Lock()
 	defer e.mu.Unlock()
-	if _, tracked := e.progress[key]; !tracked {
+	tracked, ok := e.progress[key]
+	if !ok {
 		logger.Warn("PostgreSQL apply progress discarded: the engine no longer tracks this schema change",
 			"task_id", key, "state", result.State)
 		return
 	}
-	e.progress[key] = result
+	tracked.result = result
 }
 
 func cloneMetadata(metadata map[string]string) map[string]string {
diff --git a/pkg/engine/postgres/apply_test.go b/pkg/engine/postgres/apply_test.go
index 9e6b27572..d94ff9eb3 100644
--- a/pkg/engine/postgres/apply_test.go
+++ b/pkg/engine/postgres/apply_test.go
@@ -1,17 +1,23 @@
 package postgres
 
 import (
+	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"io"
 	"log/slog"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
+	"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/jackc/pgx/v5"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -457,6 +463,451 @@ func TestProgressResultReportsCreateSequenceLength(t *testing.T) {
 	assert.Equal(t, "3", result.Metadata["steps_total"])
 }
 
+func newTestTracker(t *testing.T) *progress.Tracker {
+	t.Helper()
+	tracker, err := progress.NewTracker(progress.WallClock{})
+	require.NoError(t, err)
+	return tracker
+}
+
+// TestProgressReportsExecutorStepPosition proves a poll during execution
+// reads the step the executor is running from the pg-sprite tracker — not
+// the first-step position the engine recorded at accept — and carries that
+// step's statement in the single-line, control-free form the progress
+// metadata is contracted to hold.
+func TestProgressReportsExecutorStepPosition(t *testing.T) {
+	eng := New()
+	change := nativeApply{namespace: "public", table: "widgets", sql: "CREATE TABLE public.widgets (id bigint PRIMARY KEY)", steps: 3}
+	tracker := newTestTracker(t)
+	eng.claimProgress("task-a", progressResult(engine.StateRunning, "preflight", time.Now(), change, ""), tracker, slog.Default())
+	req := &engine.ProgressRequest{ResumeState: &engine.ResumeState{MigrationContext: "task-a"}}
+
+	before, err := eng.Progress(t.Context(), req)
+	require.NoError(t, err)
+	assert.Equal(t, "1", before.Metadata["step"], "before the executor reports, the record keeps the planned first step")
+	assert.Equal(t, "3", before.Metadata["steps_total"])
+	assert.NotContains(t, before.Metadata, "statement")
+
+	tracker.Start(3, progress.OperationAdmitting)
+	tracker.StartStep(2, progress.OperationBrief, "CREATE INDEX widgets_name_idx\n\tON public.widgets ((first_name || ' ' ||\x00 last_name))")
+
+	during, err := eng.Progress(t.Context(), req)
+	require.NoError(t, err)
+	assert.Equal(t, engine.StateRunning, during.State)
+	assert.Equal(t, "2", during.Metadata["step"])
+	assert.Equal(t, "3", during.Metadata["steps_total"])
+	assert.Equal(t, "CREATE INDEX widgets_name_idx ON public.widgets ((first_name || ' ' || last_name))", during.Metadata["statement"],
+		"the statement collapses whitespace and drops control runes but keeps its SQL operators intact")
+	assert.Equal(t, change.sql, during.Tables[0].DDL, "the table's DDL stays the planned change, not the step in flight")
+}
+
+// TestSanitizeStatementText pins the statement metadata contract: one line,
+// no control or format runes, bounded length, and otherwise the SQL exactly
+// as the executor runs it — a reason may trade its pipes for slashes because
+// it is prose, but a statement's pipes are operators.
+func TestSanitizeStatementText(t *testing.T) {
+	t.Run("keeps SQL operators and collapses layout", func(t *testing.T) {
+		got := sanitizeStatementText("SELECT 1 | 2,\n\ta || b\u202e FROM t\r\n")
+		assert.Equal(t, "SELECT 1 | 2, a || b FROM t", got)
+	})
+	t.Run("clamps on a rune boundary with an ellipsis", func(t *testing.T) {
+		long := strings.Repeat("é", maxStatementMetadataLen+40)
+		got := sanitizeStatementText(long)
+		runes := []rune(got)
+		assert.Len(t, runes, maxStatementMetadataLen)
+		assert.Equal(t, '…', runes[len(runes)-1])
+		assert.Equal(t, strings.Repeat("é", maxStatementMetadataLen-1), string(runes[:len(runes)-1]))
+	})
+	t.Run("leaves a statement at the limit untouched", func(t *testing.T) {
+		exact := strings.Repeat("x", maxStatementMetadataLen)
+		assert.Equal(t, exact, sanitizeStatementText(exact))
+	})
+}
+
+// TestPublishProgressKeepsTerminalPositionAsPublished proves a terminal
+// result answers with the position folded in at publish time: a later poll
+// never re-reads the tracker for a finished apply.
+func TestPublishProgressKeepsTerminalPositionAsPublished(t *testing.T) {
+	eng := New()
+	change := nativeApply{namespace: "public", table: "widgets", sql: "CREATE TABLE public.widgets (id bigint PRIMARY KEY)", steps: 2}
+	tracker := newTestTracker(t)
+	eng.claimProgress("task-a", progressResult(engine.StateRunning, "preflight", time.Now(), change, ""), tracker, slog.Default())
+	tracker.Start(2, progress.OperationAdmitting)
+	tracker.StartStep(2, progress.OperationBrief, "CREATE INDEX widgets_name_idx ON public.widgets (name)")
+	tracker.Finish(errors.New("boom"))
+
+	terminal := progressResult(engine.StateFailed, "failed", time.Now(), change, "detail")
+	require.NoError(t, executorProgressMetadata(t.Context(), tracker, terminal.Metadata))
+	eng.publishProgress("task-a", terminal, slog.Default())
+
+	got, err := eng.Progress(t.Context(), &engine.ProgressRequest{ResumeState: &engine.ResumeState{MigrationContext: "task-a"}})
+	require.NoError(t, err)
+	assert.Equal(t, engine.StateFailed, got.State)
+	assert.Equal(t, "2", got.Metadata["step"])
+	assert.Equal(t, "2", got.Metadata["steps_total"])
+	assert.Equal(t, "CREATE INDEX widgets_name_idx ON public.widgets (name)", got.Metadata["statement"])
+}
+
+// backgroundApplyDeadline bounds how long a unit test waits for the
+// engine's background apply drive to publish, or for a blocked call to
+// return. The drives under test run a scripted executor, so anything close
+// to this is a hang, not a slow target.
+const backgroundApplyDeadline = 10 * time.Second
+
+// scriptedExecutor stands in for pg-sprite's executor behind the engine's
+// execute seam. It hands the tracker the drive gave it to the test, then
+// waits to be released so the test can poll the engine mid-execution, and
+// returns whatever outcome the test scripted.
+type scriptedExecutor struct {
+	trackers    chan *progress.Tracker
+	released    chan struct{}
+	releaseOnce sync.Once
+	run         func(tracker *progress.Tracker) error
+}
+
+func newScriptedExecutor(run func(tracker *progress.Tracker) error) *scriptedExecutor {
+	return &scriptedExecutor{trackers: make(chan *progress.Tracker, 1), released: make(chan struct{}), run: run}
+}
+
+func (s *scriptedExecutor) execute(ctx context.Context, _ targetConn, _ nativeApply, _ int64, tracker *progress.Tracker) error {
+	s.trackers <- tracker
+	select {
+	case <-s.released:
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+	return s.run(tracker)
+}
+
+// release lets the parked executor return. It is safe to call more than
+// once, so a test can release mid-way and cleanup can release again for a
+// test that failed before it got there — otherwise the drive would stay
+// parked and Drain would wait on it for the apply ceiling.
+func (s *scriptedExecutor) release() {
+	s.releaseOnce.Do(func() { close(s.released) })
+}
+
+// tracker returns the tracker the drive handed the executor, failing the
+// test if the drive never reached it.
+func (s *scriptedExecutor) tracker(t *testing.T) *progress.Tracker {
+	t.Helper()
+	select {
+	case tracker := <-s.trackers:
+		return tracker
+	case <-time.After(backgroundApplyDeadline):
+		t.Fatal("the background drive never reached the executor")
+		return nil
+	}
+}
+
+// applyAlterUsers wires scripted in as eng's executor and accepts one
+// native-safe change through Apply under key, logging to logger, with
+// credentials the executor never dials. When the test ends the executor is
+// released and the engine drained, in that order, so a drive parked at the
+// executor cannot outlive the test or hold Drain open.
+func applyAlterUsers(t *testing.T, eng *Engine, scripted *scriptedExecutor, key string, logger *slog.Logger) {
+	t.Helper()
+	eng.execute = scripted.execute
+	t.Cleanup(eng.Drain)
+	t.Cleanup(scripted.release)
+	accepted, err := eng.Apply(t.Context(), &engine.ApplyRequest{
+		Database: "app",
+		Changes: []engine.SchemaChange{{
+			Namespace: "public",
+			TableChanges: []engine.TableChange{{
+				Table: "users", DDL: "ALTER TABLE public.users ADD COLUMN email text",
+			}},
+		}},
+		Credentials: &engine.Credentials{DSN: "postgres://schemabot:secret@db.invalid/app?sslmode=disable"},
+		ResumeState: &engine.ResumeState{MigrationContext: key},
+		Logger:      logger,
+	})
+	require.NoError(t, err)
+	require.True(t, accepted.Accepted)
+}
+
+// pollProgress reads the engine's progress for key.
+func pollProgress(t *testing.T, eng *Engine, key string) *engine.ProgressResult {
+	t.Helper()
+	got, err := eng.Progress(t.Context(), &engine.ProgressRequest{ResumeState: &engine.ResumeState{MigrationContext: key}})
+	require.NoError(t, err)
+	return got
+}
+
+// TestApplyRegistersExecutorTracker proves Apply hands the executor the very
+// tracker it claimed for the progress record, so a poll during execution
+// reads the step the executor is running, and the terminal result published
+// when the executor returns carries the position the executor finished at.
+// The poll-time tests above inject the tracker through claimProgress
+// directly; this one pins the wiring they take for granted, end to end
+// through the apply drive.
+func TestApplyRegistersExecutorTracker(t *testing.T) {
+	scripted := newScriptedExecutor(func(tracker *progress.Tracker) error {
+		tracker.Finish(nil)
+		return nil
+	})
+	eng := New()
+	const key = "task-a"
+	applyAlterUsers(t, eng, scripted, key, slog.New(slog.NewTextHandler(io.Discard, nil)))
+
+	tracker := scripted.tracker(t)
+	eng.mu.Lock()
+	tracked := eng.progress[key]
+	eng.mu.Unlock()
+	require.NotNil(t, tracked, "Apply must claim a progress record under the task identity")
+	require.Same(t, tracker, tracked.tracker, "the executor must feed the tracker the progress record reads")
+
+	tracker.Start(2, progress.OperationAdmitting)
+	tracker.StartStep(2, progress.OperationBrief, "ALTER TABLE public.users ADD COLUMN email text")
+	during := pollProgress(t, eng, key)
+	assert.Equal(t, engine.StateRunning, during.State)
+	assert.Equal(t, "2", during.Metadata["step"], "a poll mid-execution reads the executor's live step")
+	assert.Equal(t, "2", during.Metadata["steps_total"])
+
+	scripted.release()
+	require.Eventually(t, func() bool {
+		return pollProgress(t, eng, key).State.IsTerminal()
+	}, backgroundApplyDeadline, 10*time.Millisecond, "the drive must publish a terminal result once the executor returns")
+	terminal := pollProgress(t, eng, key)
+	assert.Equal(t, engine.StateCompleted, terminal.State)
+	assert.Equal(t, "2", terminal.Metadata["step"], "the terminal result carries the position the executor finished at")
+	assert.Equal(t, "2", terminal.Metadata["steps_total"])
+	assert.Equal(t, "ALTER TABLE public.users ADD COLUMN email text", terminal.Metadata["statement"])
+}
+
+// TestTerminalPublishReportsAnUnfinishedExecutorBuild pins the guard on the
+// terminal publish's premise: an executor that returns while its tracker
+// still reports a live build has broken the contract the fold-in relies on,
+// and the drive says so — the terminal result carries the tracker's
+// last-known position and the failed read is logged under the apply's
+// identifiers — instead of publishing silently as though the position were
+// final.
+func TestTerminalPublishReportsAnUnfinishedExecutorBuild(t *testing.T) {
+	session := &indexProgressSession{err: errors.New("connection reset by peer")}
+	scripted := newScriptedExecutor(func(tracker *progress.Tracker) error {
+		tracker.Start(2, progress.OperationAdmitting)
+		tracker.StartStep(2, progress.OperationConcurrentIndex, "CREATE INDEX CONCURRENTLY users_email_idx ON public.users (email)")
+		tracker.SetConcurrentBuild(session, 42)
+		return errors.New("executor returned mid-build")
+	})
+	eng := New()
+	var logs bytes.Buffer
+	const key = "task-a"
+	applyAlterUsers(t, eng, scripted, key, slog.New(slog.NewTextHandler(&logs, nil)))
+	scripted.tracker(t)
+	scripted.release()
+
+	require.Eventually(t, func() bool {
+		return pollProgress(t, eng, key).State.IsTerminal()
+	}, backgroundApplyDeadline, 10*time.Millisecond)
+	terminal := pollProgress(t, eng, key)
+	assert.Equal(t, engine.StateFailed, terminal.State)
+	assert.Equal(t, "2", terminal.Metadata["step"], "the terminal result keeps the tracker's last-known position")
+	assert.Equal(t, "CREATE INDEX CONCURRENTLY users_email_idx ON public.users (email)", terminal.Metadata["statement"])
+	assert.True(t, session.queried, "the terminal fold-in reads the tracker the executor left active")
+	assert.Contains(t, logs.String(), "terminal progress reports the last-known executor position")
+	assert.Contains(t, logs.String(), "task_id=task-a")
+	assert.Contains(t, logs.String(), "connection reset by peer")
+}
+
+// blockingProgressSession is a reserved build session whose progress read
+// parks until the test releases it, so a test can hold a tracker read open
+// and observe what else the engine lets through meanwhile.
+type blockingProgressSession struct {
+	started chan struct{}
+	release chan struct{}
+}
+
+func (s *blockingProgressSession) QueryRow(ctx context.Context, _ string, _ ...any) pgx.Row {
+	close(s.started)
+	select {
+	case <-s.release:
+	case <-ctx.Done():
+	}
+	return failingRow{err: errors.New("progress read released")}
+}
+
+// TestProgressReadsTheTrackerOutsideTheEngineLock proves a poll's tracker
+// read — a database round trip for an active concurrent index build — never
+// holds the engine lock: while one poll is parked on the server, Apply's
+// claim and the drive's terminal publish still go through. Otherwise a slow
+// progress view would stall every apply the engine tracks.
+func TestProgressReadsTheTrackerOutsideTheEngineLock(t *testing.T) {
+	eng := New()
+	change := nativeApply{namespace: "public", table: "widgets", sql: "CREATE TABLE public.widgets (id bigint PRIMARY KEY)", steps: 2}
+	tracker := newTestTracker(t)
+	tracker.Start(2, progress.OperationAdmitting)
+	tracker.StartStep(2, progress.OperationConcurrentIndex, "CREATE INDEX CONCURRENTLY widgets_name_idx ON public.widgets (name)")
+	session := &blockingProgressSession{started: make(chan struct{}), release: make(chan struct{})}
+	tracker.SetConcurrentBuild(session, 42)
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+	eng.claimProgress("task-a", progressResult(engine.StateRunning, "preflight", time.Now(), change, ""), tracker, logger)
+
+	polled := make(chan error, 1)
+	go func() {
+		_, err := eng.Progress(t.Context(), &engine.ProgressRequest{ResumeState: &engine.ResumeState{MigrationContext: "task-a"}})
+		polled <- err
+	}()
+	select {
+	case <-session.started:
+	case <-time.After(backgroundApplyDeadline):
+		t.Fatal("the poll never reached the tracker's progress read")
+	}
+
+	claimed := make(chan struct{})
+	go func() {
+		defer close(claimed)
+		eng.claimProgress("task-b", progressResult(engine.StateRunning, "preflight", time.Now(), change, ""), newTestTracker(t), logger)
+		eng.publishProgress("task-b", progressResult(engine.StateCompleted, "completed", time.Now(), change, ""), logger)
+	}()
+	select {
+	case <-claimed:
+	case <-time.After(backgroundApplyDeadline):
+		t.Fatal("a claim and publish waited behind another apply's tracker read")
+	}
+
+	close(session.release)
+	select {
+	case err := <-polled:
+		require.NoError(t, err, "a failed tracker read is logged, not returned to the poller")
+	case <-time.After(backgroundApplyDeadline):
+		t.Fatal("the parked poll did not return once its read was released")
+	}
+}
+
+// TestProgressNeverReadsTheTrackerForATerminalApply proves a poll on a
+// published terminal result answers from the record alone: even a tracker
+// that still holds an active build session is not consulted, so a finished
+// apply's poll never reaches the target.
+func TestProgressNeverReadsTheTrackerForATerminalApply(t *testing.T) {
+	eng := New()
+	change := nativeApply{namespace: "public", table: "widgets", sql: "CREATE TABLE public.widgets (id bigint PRIMARY KEY)", steps: 2}
+	tracker, session := activeBuildTracker(t, errors.New("progress view unavailable"))
+	eng.claimProgress("task-a", progressResult(engine.StateRunning, "preflight", time.Now(), change, ""), tracker, slog.Default())
+	terminal := progressResult(engine.StateFailed, "failed", time.Now(), change, "detail")
+	terminal.Metadata["step"] = "2"
+	eng.publishProgress("task-a", terminal, slog.Default())
+
+	got, err := eng.Progress(t.Context(), &engine.ProgressRequest{ResumeState: &engine.ResumeState{MigrationContext: "task-a"}})
+
+	require.NoError(t, err)
+	assert.Equal(t, engine.StateFailed, got.State)
+	assert.Equal(t, "2", got.Metadata["step"])
+	assert.False(t, session.queried, "a terminal poll must not read the tracker")
+}
+
+// TestExecutorProgressReadOutlivesTheSessionStatementTimeout pins the order
+// of the two deadlines on a tracker read: the apply pool leaves pg-sprite's
+// default statement_timeout on every session, and the engine's read deadline
+// must sit past it, so a slow progress query is cancelled by the server —
+// which hands the session back intact — before the client gives up on the
+// socket and closes the connection the build's failure verdict runs on.
+func TestExecutorProgressReadOutlivesTheSessionStatementTimeout(t *testing.T) {
+	cfg, err := spritePoolConfig("postgres://schemabot:secret@db.invalid/app", "")
+	require.NoError(t, err)
+	require.Zero(t, cfg.StatementTimeout, "the apply pool runs under pg-sprite's default statement_timeout")
+	require.Positive(t, executorProgressReadHeadroom)
+	assert.Greater(t, executorProgressReadTimeout, dbconn.DefaultStatementTimeout)
+	assert.Equal(t, dbconn.DefaultStatementTimeout+executorProgressReadHeadroom, executorProgressReadTimeout)
+}
+
+// indexProgressSession stands in for the session the executor reserves for a
+// concurrent index build, with a progress view whose read fails. It records
+// the context the read arrived on so a test can check how the engine bounds
+// it.
+type indexProgressSession struct {
+	err             error
+	queried         bool
+	cancelledAtRead error
+	hadDeadline     bool
+}
+
+func (s *indexProgressSession) QueryRow(ctx context.Context, _ string, _ ...any) pgx.Row {
+	s.queried = true
+	s.cancelledAtRead = ctx.Err()
+	_, s.hadDeadline = ctx.Deadline()
+	return failingRow{err: s.err}
+}
+
+type failingRow struct{ err error }
+
+func (r failingRow) Scan(...any) error { return r.err }
+
+// activeBuildTracker returns a tracker mid-way through a concurrent index
+// build whose server-side progress read fails with readErr.
+func activeBuildTracker(t *testing.T, readErr error) (*progress.Tracker, *indexProgressSession) {
+	t.Helper()
+	tracker := newTestTracker(t)
+	tracker.Start(2, progress.OperationAdmitting)
+	tracker.StartStep(2, progress.OperationConcurrentIndex, "CREATE INDEX CONCURRENTLY widgets_name_idx ON public.widgets (name)")
+	session := &indexProgressSession{err: readErr}
+	tracker.SetConcurrentBuild(session, 42)
+	return tracker, session
+}
+
+// TestExecutorProgressMetadataKeepsLastKnownPositionOnReadError proves the
+// read's error contract: when the server's progress view cannot be queried,
+// the metadata still receives the tracker's last-known step, total, and
+// statement, and the wrapped error goes back to the caller to log.
+func TestExecutorProgressMetadataKeepsLastKnownPositionOnReadError(t *testing.T) {
+	readErr := errors.New("connection reset by peer")
+	tracker, session := activeBuildTracker(t, readErr)
+	metadata := map[string]string{"step": "1", "steps_total": "1"}
+
+	err := executorProgressMetadata(t.Context(), tracker, metadata)
+
+	require.Error(t, err)
+	assert.ErrorIs(t, err, readErr)
+	assert.Contains(t, err.Error(), "read pg-sprite executor progress")
+	assert.True(t, session.queried)
+	assert.Equal(t, "2", metadata["step"])
+	assert.Equal(t, "2", metadata["steps_total"])
+	assert.Equal(t, "CREATE INDEX CONCURRENTLY widgets_name_idx ON public.widgets (name)", metadata["statement"])
+}
+
+// TestExecutorProgressMetadataReadsOnItsOwnBoundedContext proves the
+// progress-view read never inherits the caller's cancellation — a cancelled
+// poller or drive context must not tear down the session the executor
+// reserved for its failure verdict — while still carrying a deadline of the
+// engine's own.
+func TestExecutorProgressMetadataReadsOnItsOwnBoundedContext(t *testing.T) {
+	tracker, session := activeBuildTracker(t, errors.New("progress view unavailable"))
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+
+	err := executorProgressMetadata(ctx, tracker, map[string]string{})
+
+	require.Error(t, err)
+	require.True(t, session.queried)
+	assert.NoError(t, session.cancelledAtRead, "the read must not observe the caller's cancellation")
+	assert.True(t, session.hadDeadline, "the read must carry the engine's own deadline")
+}
+
+// TestProgressAnswersLastKnownPositionWhenTrackerReadFails proves a poll
+// whose progress-view read fails still answers: the running state and the
+// tracker's last-known position come back with a nil error, and the failure
+// is logged for triage instead of telling the driver its apply is
+// unobservable.
+func TestProgressAnswersLastKnownPositionWhenTrackerReadFails(t *testing.T) {
+	eng := New()
+	change := nativeApply{namespace: "public", table: "widgets", sql: "CREATE TABLE public.widgets (id bigint PRIMARY KEY)", steps: 2}
+	tracker, _ := activeBuildTracker(t, errors.New("connection reset by peer"))
+	var logs bytes.Buffer
+	logger := slog.New(slog.NewTextHandler(&logs, nil))
+	eng.claimProgress("task-a", progressResult(engine.StateRunning, "preflight", time.Now(), change, ""), tracker, logger)
+
+	got, err := eng.Progress(t.Context(), &engine.ProgressRequest{ResumeState: &engine.ResumeState{MigrationContext: "task-a"}})
+
+	require.NoError(t, err)
+	assert.Equal(t, engine.StateRunning, got.State)
+	assert.Equal(t, "2", got.Metadata["step"])
+	assert.Equal(t, "2", got.Metadata["steps_total"])
+	assert.Equal(t, "CREATE INDEX CONCURRENTLY widgets_name_idx ON public.widgets (name)", got.Metadata["statement"])
+	assert.Contains(t, logs.String(), "reports the last-known executor position")
+	assert.Contains(t, logs.String(), "task_id=task-a")
+	assert.Contains(t, logs.String(), "connection reset by peer")
+}
+
 // TestRefusalForOutcomeTotalOverExecutorCodes pins the classifier to
 // pg-sprite's full outcome vocabulary: every code the executor can return
 // maps to an explicit disposition — refusal or operational — so a code added
@@ -628,7 +1079,7 @@ func TestInvalidIndexDetailMatchesVerdictOwnership(t *testing.T) {
 func TestProgressIsKeyedToTheRequestingApply(t *testing.T) {
 	eng := New()
 	change := nativeApply{namespace: "public", table: "t_a", sql: "ALTER TABLE public.t_a ADD COLUMN a text"}
-	eng.claimProgress("task-a", progressResult(engine.StateCompleted, "completed", time.Now(), change, ""))
+	eng.claimProgress("task-a", progressResult(engine.StateCompleted, "completed", time.Now(), change, ""), newTestTracker(t), slog.Default())
 
 	tracked, err := eng.Progress(t.Context(), &engine.ProgressRequest{
 		ResumeState: &engine.ResumeState{MigrationContext: "task-a"},
@@ -660,8 +1111,8 @@ func TestConcurrentAppliesEachAnswerForTheirOwnWork(t *testing.T) {
 	eng := New()
 	changeA := nativeApply{namespace: "public", table: "t_a", sql: "ALTER TABLE public.t_a ADD COLUMN a text"}
 	changeB := nativeApply{namespace: "public", table: "t_b", sql: "ALTER TABLE public.t_b ADD COLUMN b text"}
-	eng.claimProgress("task-a", progressResult(engine.StateRunning, "preflight", time.Now(), changeA, ""))
-	eng.claimProgress("task-b", progressResult(engine.StateRunning, "preflight", time.Now(), changeB, ""))
+	eng.claimProgress("task-a", progressResult(engine.StateRunning, "preflight", time.Now(), changeA, ""), newTestTracker(t), slog.Default())
+	eng.claimProgress("task-b", progressResult(engine.StateRunning, "preflight", time.Now(), changeB, ""), newTestTracker(t), slog.Default())
 
 	first, err := eng.Progress(t.Context(), &engine.ProgressRequest{
 		ResumeState: &engine.ResumeState{MigrationContext: "task-a"},
@@ -689,10 +1140,10 @@ func TestClaimProgressRetiresSettledApplies(t *testing.T) {
 	settled := nativeApply{namespace: "public", table: "t_settled", sql: "ALTER TABLE public.t_settled ADD COLUMN a text"}
 	running := nativeApply{namespace: "public", table: "t_running", sql: "ALTER TABLE public.t_running ADD COLUMN b text"}
 	fresh := nativeApply{namespace: "public", table: "t_fresh", sql: "ALTER TABLE public.t_fresh ADD COLUMN c text"}
-	eng.claimProgress("task-settled", progressResult(engine.StateCompleted, "completed", time.Now(), settled, ""))
-	eng.claimProgress("task-running", progressResult(engine.StateRunning, "preflight", time.Now(), running, ""))
+	eng.claimProgress("task-settled", progressResult(engine.StateCompleted, "completed", time.Now(), settled, ""), newTestTracker(t), slog.Default())
+	eng.claimProgress("task-running", progressResult(engine.StateRunning, "preflight", time.Now(), running, ""), newTestTracker(t), slog.Default())
 
-	eng.claimProgress("task-fresh", progressResult(engine.StateRunning, "preflight", time.Now(), fresh, ""))
+	eng.claimProgress("task-fresh", progressResult(engine.StateRunning, "preflight", time.Now(), fresh, ""), newTestTracker(t), slog.Default())
 
 	retired, err := eng.Progress(t.Context(), &engine.ProgressRequest{
 		ResumeState: &engine.ResumeState{MigrationContext: "task-settled"},
@@ -718,7 +1169,7 @@ func TestUntrackedApplyProgressIsDiscarded(t *testing.T) {
 	eng := New()
 	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
 	changeB := nativeApply{namespace: "public", table: "t_b", sql: "ALTER TABLE public.t_b ADD COLUMN b text"}
-	eng.claimProgress("task-b", progressResult(engine.StateRunning, "preflight", time.Now(), changeB, ""))
+	eng.claimProgress("task-b", progressResult(engine.StateRunning, "preflight", time.Now(), changeB, ""), newTestTracker(t), slog.Default())
 
 	changeA := nativeApply{namespace: "public", table: "t_a", sql: "ALTER TABLE public.t_a ADD COLUMN a text"}
 	eng.publishProgress("task-a", progressResult(engine.StateCompleted, "completed", time.Now(), changeA, ""), logger)
@@ -746,8 +1197,8 @@ func TestDrainStopsTrackingEverySchemaChange(t *testing.T) {
 	eng := New()
 	changeA := nativeApply{namespace: "public", table: "t_a", sql: "ALTER TABLE public.t_a ADD COLUMN a text"}
 	changeB := nativeApply{namespace: "public", table: "t_b", sql: "ALTER TABLE public.t_b ADD COLUMN b text"}
-	eng.claimProgress("task-a", progressResult(engine.StateCompleted, "completed", time.Now(), changeA, ""))
-	eng.claimProgress("task-b", progressResult(engine.StateRunning, "preflight", time.Now(), changeB, ""))
+	eng.claimProgress("task-a", progressResult(engine.StateCompleted, "completed", time.Now(), changeA, ""), newTestTracker(t), slog.Default())
+	eng.claimProgress("task-b", progressResult(engine.StateRunning, "preflight", time.Now(), changeB, ""), newTestTracker(t), slog.Default())
 
 	eng.Drain()
 
@@ -821,7 +1272,7 @@ func TestExecuteOptimisticRefusesUnreadableCABundle(t *testing.T) {
 		caCertPath: filepath.Join(t.TempDir(), "missing.pem"),
 	}
 
-	err := executeOptimistic(t.Context(), conn, nativeApply{namespace: "public", table: "widgets", sql: "CREATE TABLE widgets (id bigint PRIMARY KEY)"}, DefaultNativeSafeTableSizeLimitBytes)
+	err := executeOptimistic(t.Context(), conn, nativeApply{namespace: "public", table: "widgets", sql: "CREATE TABLE widgets (id bigint PRIMARY KEY)"}, DefaultNativeSafeTableSizeLimitBytes, newTestTracker(t))
 
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "open pg-sprite apply pool")
diff --git a/pkg/engine/postgres/postgres.go b/pkg/engine/postgres/postgres.go
index 7cda0f0e9..62d7644bc 100644
--- a/pkg/engine/postgres/postgres.go
+++ b/pkg/engine/postgres/postgres.go
@@ -18,6 +18,7 @@ import (
 	pgplan "github.com/block/pg-sprite/pkg/plan"
 	"github.com/block/pg-sprite/pkg/planner"
 	"github.com/block/pg-sprite/pkg/preflight"
+	"github.com/block/pg-sprite/pkg/progress"
 	"github.com/block/pg-sprite/pkg/router"
 	pgstatement "github.com/block/pg-sprite/pkg/statement"
 	spirittable "github.com/block/spirit/pkg/table"
@@ -44,8 +45,26 @@ type Engine struct {
 	// accepting a second apply on the same target must not evict the first
 	// one's state while it is still running, or the running apply's driver
 	// would be told its work no longer exists.
-	progress       map[string]*engine.ProgressResult
+	progress       map[string]*trackedApply
 	tableSizeLimit int64
+
+	// execute is a test seam standing in for executeOptimistic, so the apply
+	// drive — accept, claim, execute, terminal publish — can be exercised
+	// against an executor the test scripts instead of a target to dial. Nil
+	// selects the real executor.
+	execute func(ctx context.Context, conn targetConn, change nativeApply, tableSizeLimit int64, tracker *progress.Tracker) error
+}
+
+// trackedApply pairs the progress the engine has published for one apply
+// with the pg-sprite tracker its executor feeds. The published result changes
+// only at accept and at the terminal outcome; the tracker is what moves in
+// between, so Progress reads the step position and statement from it. The
+// logger is the apply's own, so a poll that cannot read the tracker logs
+// under the identifiers the apply was accepted with.
+type trackedApply struct {
+	result  *engine.ProgressResult
+	tracker *progress.Tracker
+	logger  *slog.Logger
 }
 
 // DefaultNativeSafeTableSizeLimitBytes preserves the native-safe execution
@@ -648,14 +667,45 @@ func blockChangesAtTier(changes []engine.TableChange, tiers []preflight.Tier, ti
 // and the table cell separator is neutralized so a crafted identifier cannot
 // break comment layout.
 func sanitizeReasonText(s string) string {
+	return strings.ReplaceAll(singleLine(s), "|", "/")
+}
+
+// maxStatementMetadataLen bounds the statement text carried in progress
+// metadata. The value is stored and rendered alongside other clamped
+// operator-facing summaries, and a statement is unbounded input — a create
+// set's index definition can run to any length — so it is cut on a rune
+// boundary with an ellipsis rather than trusted to fit.
+const maxStatementMetadataLen = 255
+
+// sanitizeStatementText prepares the SQL the executor is running for
+// progress metadata. Unlike a reason, which is prose SchemaBot composes, a
+// statement is quoted back to the operator as the SQL it is: control and
+// format characters are stripped and whitespace collapses to one line, but
+// the text is otherwise left as written, so `||` stays concatenation and
+// `1 | 2` stays a bitwise or. A surface that embeds the value in Markdown
+// backslash-escapes the delimiters it cares about at render time, the way
+// the comment templates already do for engine-influenced inline text; the
+// metadata carries the statement, not one surface's escaping of it.
+func sanitizeStatementText(s string) string {
+	s = singleLine(s)
+	runes := []rune(s)
+	if len(runes) > maxStatementMetadataLen {
+		return string(runes[:maxStatementMetadataLen-1]) + "…"
+	}
+	return s
+}
+
+// singleLine strips control and format characters and collapses every
+// whitespace run — newlines included — to one space, so the result cannot
+// span lines or carry a bidi override.
+func singleLine(s string) string {
 	s = strings.Map(func(r rune) rune {
 		if unicode.IsControl(r) || unicode.Is(unicode.Cf, r) {
 			return ' '
 		}
 		return r
 	}, s)
-	s = strings.Join(strings.Fields(s), " ")
-	return strings.ReplaceAll(s, "|", "/")
+	return strings.Join(strings.Fields(s), " ")
 }
 
 func executionVerdict(formatVersion int, statement pgplan.Statement, table string) (string, string) {
diff --git a/pkg/engine/postgres/postgres_integration_test.go b/pkg/engine/postgres/postgres_integration_test.go
index e6fee053f..48ff04029 100644
--- a/pkg/engine/postgres/postgres_integration_test.go
+++ b/pkg/engine/postgres/postgres_integration_test.go
@@ -621,6 +621,9 @@ func TestEngineApplyNativeSafe(t *testing.T) {
 	assert.Equal(t, 100, progress.Progress)
 	assert.Equal(t, "completed", progress.Metadata["phase"])
 	assert.Equal(t, "1", progress.Metadata["step"])
+	assert.Equal(t, "1", progress.Metadata["steps_total"])
+	assert.Equal(t, "ALTER TABLE public.users ADD COLUMN email text", progress.Metadata["statement"],
+		"the terminal position names the statement the executor ran")
 
 	var exists bool
 	err = db.QueryRowContext(t.Context(), `SELECT EXISTS (
@@ -681,6 +684,9 @@ func TestEngineApplyGreenfieldCreateSet(t *testing.T) {
 	assert.True(t, result.Accepted)
 	progress := awaitPostgresProgress(t, eng, "widgets")
 	assert.Equal(t, engine.StateCompleted, progress.State)
+	assert.Equal(t, "3", progress.Metadata["step"], "a completed create set reports its last step, not the first")
+	assert.Equal(t, "3", progress.Metadata["steps_total"])
+	assert.Contains(t, progress.Metadata["statement"], "CREATE INDEX widgets_id_idx")
 
 	rows, err := db.QueryContext(t.Context(), "SELECT indexname FROM pg_indexes WHERE schemaname = 'public' AND tablename = 'widgets' ORDER BY indexname")
 	require.NoError(t, err)
@@ -753,6 +759,9 @@ func TestEngineApplyCreateSetCommittedPrefixNotRetryable(t *testing.T) {
 	assert.Equal(t, "failed", progress.Metadata["phase"])
 	assert.False(t, progress.Retryable, "the CREATE TABLE committed; a retry cannot succeed, so the drive must not offer one")
 	assert.Equal(t, `step 2 of 2 failed after the CREATE TABLE for "widgets" committed; re-plan against the current schema`, progress.ErrorMessage)
+	assert.Equal(t, "2", progress.Metadata["step"], "the position names the step that failed")
+	assert.Equal(t, "2", progress.Metadata["steps_total"])
+	assert.Contains(t, progress.Metadata["statement"], "CREATE INDEX widgets_name_idx")
 
 	var exists bool
 	require.NoError(t, db.QueryRowContext(t.Context(), "SELECT to_regclass('public.widgets') IS NOT NULL").Scan(&exists))
diff --git a/pkg/tern/local_apply.go b/pkg/tern/local_apply.go
index dfd20c2df..7bdcc4630 100644
--- a/pkg/tern/local_apply.go
+++ b/pkg/tern/local_apply.go
@@ -58,8 +58,12 @@ func (c *LocalClient) checkActiveTaskConflict(ctx context.Context, plan *storage
 			return blockingTask{}, released, nil
 		}
 
-		// Retry: 10 attempts with 100ms sleep gives 1 second total wait.
-		// Handles the race where storage is updated but Spirit hasn't fully finished.
+		// Retry a bounded number of times with a short sleep between
+		// attempts, to ride out the window where storage is updated but the
+		// engine has not fully finished. The sleep is the floor of each
+		// attempt, not its length: an attempt that probes a running apply
+		// waits on the engine's progress read, so the loop's worst case is
+		// the attempt count times that read's own bound.
 		if attempt < 9 {
 			c.logger.Debug("found potentially stale active task, retrying",
 				"task_id", blocking.taskIdentifier, "table", blocking.table, "shard", blocking.shard,
@@ -576,8 +580,8 @@ func (c *LocalClient) pendingDriverRequest(ctx context.Context, apply *storage.A
 // storage believes is in-flight, the task is updated in storage and no longer blocks.
 // Resting tasks (Stopped, FailedRetryable) are left untouched.
 //
-// The engine probe is in-memory and database-scoped: it reports this process's
-// last run on the database, not the task's actual cross-process state. The
+// The engine probe answers from this process's own memory of the work: it
+// reports what this process ran, not the task's actual cross-process state. The
 // task's parent apply lease decides whether that memory is authoritative — a
 // fresh lease means a live driver owns the work and the task keeps blocking,
 // and a terminal report is only trusted when the last lease belongs to this
@@ -597,12 +601,17 @@ func (c *LocalClient) tryResolveStaleTask(ctx context.Context, t *storage.Task,
 		return false
 	}
 
-	// The raw target credentials (no namespace mapping) are safe here only
-	// because Spirit's Progress is purely in-memory and never queries by
-	// request database or connection schema. An engine whose Progress inspects
-	// the database must resolve credentials per task (credentialsForTask)
-	// before this probe, or under schema overrides it would address the
-	// canonical name instead of the physical schema.
+	// The raw target credentials (no namespace mapping) are correct here
+	// because per-namespace resolution only exists for MySQL, whose engine
+	// never connects from a Progress request — Spirit's progress is purely
+	// in-memory. For every other database type credentialsForTask is the
+	// identity: the target-level credentials are the task's credentials, so
+	// an engine whose Progress does connect from them (PlanetScale builds an
+	// API client; PostgreSQL reads only through the session its own executor
+	// already holds) addresses exactly what the apply itself addressed. Were
+	// MySQL's engine ever to connect from these credentials, this probe would
+	// have to resolve them per task first, or under schema overrides it
+	// would address the canonical name instead of the physical schema.
 	//
 	// The task identifier rides along for engines that key progress by apply
 	// identity (postgres): a probe about work the engine is still running