feat(github): render live row-copy progress on sharded table lines

TEMPLATES.md

@@ -7691,8 +7691,9 @@ schemabot apply -e production
 
 #### Keyspace `cdb_resolute_sharded`
 
-**`mutes`**: 🔄 Row copy in progress
-  └ shards: ◐ -40 45% · ⏳ 40-80 · ⏳ 80-c0 · ⏳ c0-
+**`mutes`**: 🟦🟦🟦🟦🟦🟦🟦🟦🟦🟦🟦🟦⬜⬜⬜⬜⬜⬜⬜⬜ 62% (1 of 4 shards)
+- Rows: 914,707 / 1,466,232 across 1 of 4 shards · ETA: ≥ 3m 15s
+  └ shards: ◐ -40 62% · ⏳ 40-80 · ⏳ 80-c0 · ⏳ c0-
 
 _Last updated: <relative-time datetime="2026-01-01T00:00:00Z">2026-01-01 00:00:00 UTC</relative-time> (2026-01-01 00:00:00 UTC)_
 
@@ -7792,7 +7793,8 @@ _Last updated: <relative-time datetime="2026-01-01T00:00:00Z">2026-01-01 00:00:0
 
 #### Keyspace `cdb_resolute_lookup`
 
-**`outcomes_lookup`**: 🔄 Row copy in progress
+**`outcomes_lookup`**: 🟦🟦🟦🟦🟦⬜⬜⬜⬜⬜⬜⬜⬜⬜⬜⬜⬜⬜⬜⬜ 27%
+- Rows: 540,211 / 2,000,780 · ETA: 8m 0s
 
 #### Keyspace `cdb_resolute_sharded`
 
@@ -7906,8 +7908,8 @@ schemabot apply -e production
 | --- | --- |
 | `-40` | ✅ completed |
 | `40-80` | ✅ completed |
-| `80-c0` | ⛔ cancelled |
-| `c0-` | ⛔ cancelled |
+| `80-c0` | 🚫 cancelled |
+| `c0-` | 🚫 cancelled |
 
 ---
 

pkg/webhook/sharded_apply.go

@@ -363,61 +363,103 @@ func shardStatusesByKeyspace(groups []shardWorkGroup, qualifyIdentity bool, rele
 // siblings copy.
 func shardedTableStatusesByKeyspace(ops []*storage.ApplyOperation, tasksByOp map[int64][]*storage.Task) map[string][]templates.ShardedTableStatus {
 	type keyspaceTable struct{ namespace, table string }
-	seen := make(map[keyspaceTable]struct{})
+	type tableRollup struct {
+		shards          []templates.ShardProgressData
+		rowsCopied      int64
+		rowsTotal       int64
+		etaSeconds      int64
+		shardsReporting int
+	}
 	var order []keyspaceTable
-	shardsByTable := make(map[keyspaceTable][]templates.ShardProgressData)
+	rollups := make(map[keyspaceTable]*tableRollup)
 	for _, op := range ops {
 		ns, shard, table, ok := parseShardOperationKey(op.OperationKey)
 		if !ok {
 			// Finalizers render in the VSchema section, not as a table.
 			continue
 		}
 		key := keyspaceTable{namespace: ns, table: table}
-		if _, dup := seen[key]; !dup {
-			seen[key] = struct{}{}
+		r := rollups[key]
+		if r == nil {
+			r = &tableRollup{}
+			rollups[key] = r
 			order = append(order, key)
 		}
-		status, percent := shardTaskStatus(op, tasksByOp[op.ID])
-		shardsByTable[key] = append(shardsByTable[key], templates.ShardProgressData{
+		sp := shardTaskProgress(op, tasksByOp[op.ID])
+		r.shards = append(r.shards, templates.ShardProgressData{
 			Shard:           shard,
-			Status:          status,
-			PercentComplete: percent,
+			Status:          sp.status,
+			PercentComplete: sp.percent,
 		})
+		// Rows sum across the shards that have reported; the ETA is the slowest
+		// reporting shard's. A shard counts as reporting only once it carries a
+		// row total, and all of its figures are gated on that together — the
+		// numerator, denominator, ETA, and coverage count always describe the
+		// same set of shards, so a shard with copied rows but no total yet
+		// cannot inflate the fraction's numerator alone. Shards whose dispatch
+		// wave has not started contribute nothing, and the renderer discloses
+		// the coverage instead of presenting a wave's figures as the table's.
+		if sp.rowsTotal > 0 {
+			r.shardsReporting++
+			r.rowsCopied += sp.rowsCopied
+			r.rowsTotal += sp.rowsTotal
+			if sp.etaSeconds > r.etaSeconds {
+				r.etaSeconds = sp.etaSeconds
+			}
+		}
 	}
 	out := make(map[string][]templates.ShardedTableStatus, len(order))
 	for _, key := range order {
-		shards := shardsByTable[key]
+		r := rollups[key]
 		out[key.namespace] = append(out[key.namespace], templates.ShardedTableStatus{
-			Table:  key.table,
-			Status: aggregateTableStatus(shards),
-			Shards: shards,
+			Table:           key.table,
+			Status:          aggregateTableStatus(r.shards),
+			RowsCopied:      r.rowsCopied,
+			RowsTotal:       r.rowsTotal,
+			ETASeconds:      r.etaSeconds,
+			ShardsReporting: r.shardsReporting,
+			Shards:          r.shards,
 		})
 	}
 	return out
 }
 
-// shardTaskStatus resolves one (shard, table) operation's display status and
-// copy percent from its most attention-worthy task — the task is where the
+// shardProgress is one (shard, table) operation's display projection: the
+// state and copy figures of its most attention-worthy task.
+type shardProgress struct {
+	status     string
+	percent    int
+	rowsCopied int64
+	rowsTotal  int64
+	etaSeconds int64
+}
+
+// shardTaskProgress resolves one (shard, table) operation's display status and
+// copy figures from its most attention-worthy task — the task is where the
 // engine reports live shard state. The operation state stands in when the
 // operation has no tasks yet (dispatch creates them when its wave starts) or a
 // task has not reported state; it normalizes into the same vocabulary.
-func shardTaskStatus(op *storage.ApplyOperation, tasks []*storage.Task) (string, int) {
-	best := ""
-	percent := 0
+func shardTaskProgress(op *storage.ApplyOperation, tasks []*storage.Task) shardProgress {
+	best := shardProgress{}
 	for _, t := range tasks {
 		status := t.State
 		if status == "" {
 			status = op.State
 		}
-		if best == "" || taskStateRank(status) > taskStateRank(best) {
-			best = status
-			percent = t.ProgressPercent
+		if best.status == "" || taskStateRank(status) > taskStateRank(best.status) {
+			best = shardProgress{
+				status:     status,
+				percent:    t.ProgressPercent,
+				rowsCopied: t.RowsCopied,
+				rowsTotal:  t.RowsTotal,
+				etaSeconds: int64(t.ETASeconds),
+			}
 		}
 	}
-	if best == "" {
-		return op.State, 0
+	if best.status == "" {
+		return shardProgress{status: op.State}
 	}
-	return best, percent
+	return best
 }
 
 // aggregateTableStatus reduces a table's per-shard states to the one an

pkg/webhook/sharded_apply_test.go

@@ -570,17 +570,18 @@ func TestBuildShardedApplyData_TableRollupFromTasks(t *testing.T) {
 		mk(2, "cdb_resolute_sharded/40-80/mutes", state.ApplyOperation.Running),
 		mk(3, "cdb_resolute_sharded/80-/mutes", state.ApplyOperation.Pending),
 	}
-	task := func(id, opID int64, shard, taskState string, percent int) *storage.Task {
+	task := func(id, opID int64, shard, taskState string, percent int, copied, total int64, eta int) *storage.Task {
 		oid := opID
 		return &storage.Task{
 			ID: id, ApplyID: 1, ApplyOperationID: &oid, Shard: shard,
 			Namespace: "cdb_resolute_sharded", TableName: "mutes",
 			State: taskState, ProgressPercent: percent,
+			RowsCopied: copied, RowsTotal: total, ETASeconds: eta,
 		}
 	}
 	tasks := []*storage.Task{
-		task(1, 1, "-40", state.Task.Completed, 100),
-		task(2, 2, "40-80", state.Task.Running, 37),
+		task(1, 1, "-40", state.Task.Completed, 100, 500000, 500000, 0),
+		task(2, 2, "40-80", state.Task.Running, 37, 185000, 500000, 240),
 		// The 80- operation has no task yet: dispatch creates tasks when the
 		// shard's wave starts, so its operation state stands in.
 	}
@@ -598,6 +599,42 @@ func TestBuildShardedApplyData_TableRollupFromTasks(t *testing.T) {
 	assert.Equal(t, templates.ShardProgressData{Shard: "40-80", Status: state.Task.Running, PercentComplete: 37}, table.Shards[1])
 	assert.Equal(t, templates.ShardProgressData{Shard: "80-", Status: state.ApplyOperation.Pending, PercentComplete: 0}, table.Shards[2],
 		"an operation without a task contributes its operation state")
+	assert.Equal(t, int64(685000), table.RowsCopied, "rows sum across the shards that have reported")
+	assert.Equal(t, int64(1000000), table.RowsTotal, "the taskless shard contributes no rows yet")
+	assert.Equal(t, int64(240), table.ETASeconds, "the ETA is the slowest reporting shard's")
+	assert.Equal(t, 2, table.ShardsReporting, "the taskless shard is not counted as reporting")
+}
+
+// A shard reporting copied rows without a row total has no denominator to
+// aggregate against, so none of its figures count: the table's fraction stays
+// consistent (numerator, denominator, ETA, and coverage all describe the same
+// reporting shards) instead of copied rows inflating the numerator alone.
+func TestBuildShardedApplyData_CopiedRowsWithoutTotalNotAggregated(t *testing.T) {
+	mk := func(id int64, key, opState string) *storage.ApplyOperation {
+		return &storage.ApplyOperation{ID: id, ApplyID: 1, Deployment: "cake", OperationKey: key, State: opState, CutoverPolicy: storage.CutoverPolicyRolling, OnFailure: storage.OnFailureHalt}
+	}
+	ops := []*storage.ApplyOperation{
+		mk(1, "cdb_resolute_sharded/-40/mutes", state.ApplyOperation.Running),
+		mk(2, "cdb_resolute_sharded/40-/mutes", state.ApplyOperation.Running),
+	}
+	opID1, opID2 := int64(1), int64(2)
+	tasks := []*storage.Task{
+		{ID: 1, ApplyID: 1, ApplyOperationID: &opID1, Shard: "-40", Namespace: "cdb_resolute_sharded", TableName: "mutes",
+			State: state.Task.Running, ProgressPercent: 37, RowsCopied: 185000, RowsTotal: 500000, ETASeconds: 240},
+		{ID: 2, ApplyID: 1, ApplyOperationID: &opID2, Shard: "40-", Namespace: "cdb_resolute_sharded", TableName: "mutes",
+			State: state.Task.Running, RowsCopied: 90000, ETASeconds: 900},
+	}
+	apply := &storage.Apply{ApplyIdentifier: "apply-x", Database: "cdb_resolute", Environment: "staging", State: state.Apply.Running}
+
+	data := buildShardedApplyData(apply, ops, false, tasks, nil, "")
+
+	require.Len(t, data.Keyspaces, 1)
+	require.Len(t, data.Keyspaces[0].Tables, 1)
+	table := data.Keyspaces[0].Tables[0]
+	assert.Equal(t, int64(185000), table.RowsCopied, "copied rows without a total stay out of the numerator")
+	assert.Equal(t, int64(500000), table.RowsTotal)
+	assert.Equal(t, int64(240), table.ETASeconds, "an ETA without a total does not set the table's floor")
+	assert.Equal(t, 1, table.ShardsReporting, "a shard without a row total is not reporting")
 }
 
 // A shard whose table failed makes the whole table read failed, and each

pkg/webhook/templates/preview_sharded.go

@@ -24,6 +24,13 @@ func previewShardStatuses(ops []presentation.Operation) []ShardStatus {
 const (
 	previewMutesIndex      = "ALTER TABLE `mutes` ADD INDEX `created_at`(`created_at`);"
 	previewMutesIndexDrift = "ALTER TABLE `mutes` ADD INDEX `created_at`(`created_at`), ADD COLUMN `reason` varchar(255);"
+
+	// Copy figures for the mutes fixture's reporting shard. The shard percent
+	// is derived from the row figures so the summary line and the rows line
+	// cannot drift apart when either is edited.
+	previewMutesRowsCopied  = int64(914707)
+	previewMutesRowsTotal   = int64(1466232)
+	previewMutesCopyPercent = int(previewMutesRowsCopied * 100 / previewMutesRowsTotal)
 )
 
 func previewMutesCell(shard string) ShardCell {
@@ -40,8 +47,10 @@ func PreviewCommentShardedApplyInProgress() string {
 			Keyspace: "cdb_resolute_sharded",
 			Tables: []ShardedTableStatus{{
 				Table: "mutes", Status: state.Task.Running,
+				RowsCopied: previewMutesRowsCopied, RowsTotal: previewMutesRowsTotal, ETASeconds: 195,
+				ShardsReporting: 1,
 				Shards: []ShardProgressData{
-					{Shard: "-40", Status: state.Task.Running, PercentComplete: 45},
+					{Shard: "-40", Status: state.Task.Running, PercentComplete: previewMutesCopyPercent},
 					{Shard: "40-80", Status: state.Task.Pending},
 					{Shard: "80-c0", Status: state.Task.Pending},
 					{Shard: "c0-", Status: state.Task.Pending},
@@ -214,7 +223,7 @@ func PreviewCommentShardedApplyDivergent() string {
 			Tables: []ShardedTableStatus{{
 				Table: "mutes", Status: state.Task.Running,
 				Shards: []ShardProgressData{
-					{Shard: "-40", Status: state.Task.Running, PercentComplete: 62},
+					{Shard: "-40", Status: state.Task.Running, PercentComplete: previewMutesCopyPercent},
 					{Shard: "40-80", Status: state.Task.Pending},
 					{Shard: "80-c0", Status: state.Task.Pending},
 				},
@@ -269,7 +278,9 @@ func PreviewCommentShardedApplyMultiKeyspace() string {
 				Keyspace: "cdb_resolute_lookup",
 				Tables: []ShardedTableStatus{{
 					Table: "outcomes_lookup", Status: state.Task.Running,
-					Shards: []ShardProgressData{{Shard: "-", Status: state.Task.Running, PercentComplete: 27}},
+					RowsCopied: 540211, RowsTotal: 2000780, ETASeconds: 480,
+					ShardsReporting: 1,
+					Shards:          []ShardProgressData{{Shard: "-", Status: state.Task.Running, PercentComplete: 27}},
 				}},
 				Shards: []ShardStatus{unshard(shards[1], "-")},
 				Cells:  []ShardCell{{Shard: "-", Table: "outcomes_lookup", DDL: "ALTER TABLE `outcomes_lookup` ADD COLUMN `verdict` varchar(32);"}},