From 7b303954f353468ce1786305673ba0820ad8d84d Mon Sep 17 00:00:00 2001
From: Armand Parajon <armand@squareup.com>
Date: Thu, 13 Aug 2026 18:50:47 +0800
Subject: [PATCH 1/6] feat(observability): make a resumed drive explain itself
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two gaps left an apply silent for the drive that actually ran it:

- The engine's log callback was wired only by the drive that started an
  apply, so the apply log stream went quiet the moment the apply changed
  hands. Both resume paths now wire it for as long as they drive engine
  work; a detached grouped resume unwires it from the goroutine that
  outlives the call.
- Every resume announced itself as a heartbeat expiry, a cause this path
  never checks — the claim arm that selected the apply is decided by the
  operator's claim query and is not carried here. It now reports what it
  knows and leaves the cause to the claim logs.
---
 pkg/tern/local_control_resume.go              |  32 +++++-
 ..._resume_engine_logging_integration_test.go | 106 ++++++++++++++++++
 2 files changed, 135 insertions(+), 3 deletions(-)
 create mode 100644 pkg/tern/local_resume_engine_logging_integration_test.go

diff --git a/pkg/tern/local_control_resume.go b/pkg/tern/local_control_resume.go
index 80f466f1e..20280b017 100644
--- a/pkg/tern/local_control_resume.go
+++ b/pkg/tern/local_control_resume.go
@@ -290,6 +290,11 @@ func (c *LocalClient) resumeApplySequential(ctx context.Context, apply *storage.
 	ctx, cancelApply := context.WithCancel(ctx)
 	defer cancelApply()
 	defer c.startApplyHeartbeat(ctx, apply, cancelApply)()
+	// A resumed drive runs the same engine work as the drive that started it, so
+	// it needs the same log wiring: without it the engine's own lines stop
+	// reaching the apply log stream the moment an apply changes hands, and the
+	// stream goes quiet for exactly the drive an operator is trying to read.
+	defer c.setupSpiritLogging(ctx, apply, tasks)()
 	creds := c.credentials()
 	eng := c.getEngine()
 	// Bind the apply's identity once so every line of this sequential resume is
@@ -764,6 +769,20 @@ func (c *LocalClient) launchAtomicResume(ctx context.Context, apply *storage.App
 		return err
 	}
 
+	// Route the engine's own log lines into this apply's log stream, so a
+	// resumed drive reads like the drive that started it. The wiring goes up
+	// before the engine accepts the resume — it emits setup and lock lines from
+	// inside Apply — and comes down when the drive that polls the work returns.
+	// A detached resume polls in its own goroutine that outlives this call, so
+	// that goroutine unwires it instead.
+	stopEngineLogging := c.setupSpiritLogging(ctx, apply, tasks)
+	pollDetached := false
+	defer func() {
+		if !pollDetached {
+			stopEngineLogging()
+		}
+	}()
+
 	// Resume the grouped apply with the engine's persisted state so it
 	// reattaches to in-flight engine work instead of launching a duplicate
 	// schema change. The changes are rebuilt from the stored tasks so the
@@ -835,9 +854,11 @@ func (c *LocalClient) launchAtomicResume(ctx context.Context, apply *storage.App
 
 	resumeCtx, cancelResume := context.WithCancel(context.WithoutCancel(ctx))
 	stopHeartbeat := c.startParentApplyHeartbeat(resumeCtx, apply, suppressParent, cancelResume)
+	pollDetached = true
 	go func() {
 		defer cancelResume()
 		defer stopHeartbeat()
+		defer stopEngineLogging()
 		c.pollForCompletionAtomic(resumeCtx, apply, tasks, creds, resumeState, options, releaseAtCutoverBarrier)
 	}()
 	return nil
@@ -1510,14 +1531,19 @@ func (c *LocalClient) resumeApplyWithTasks(ctx context.Context, apply *storage.A
 		return ctx.Err()
 	}
 
-	logger.Info("resuming apply (heartbeat expired)",
+	// This drive knows only that it holds the claim on an already-started apply;
+	// the claim arm that selected it — a stale heartbeat, a pending control
+	// request, a barrier-parked cutover — is decided by the operator's claim
+	// query and is not carried here. Report what is known and let the operator's
+	// claim logs name the cause, rather than asserting one this path never
+	// checked.
+	logger.Info("recovering apply: resuming an already-started apply from its stored state",
 		"state", apply.State,
 		"task_count", len(tasks),
 	)
 
-	// Log recovery event
 	c.logApplyEvent(ctx, apply.ID, nil, storage.LogLevelInfo, storage.LogEventInfo, storage.LogSourceSchemaBot,
-		fmt.Sprintf("Recovering apply (heartbeat expired, was in %s state)", apply.State), "", "")
+		fmt.Sprintf("Recovering apply from its stored state (was in %s state)", apply.State), "", "")
 
 	deferredCutoverSignalAbsent := false
 	if shouldInspectCutoverSignalForResume(apply, forceCutoverResume) {
diff --git a/pkg/tern/local_resume_engine_logging_integration_test.go b/pkg/tern/local_resume_engine_logging_integration_test.go
new file mode 100644
index 000000000..3bb9ea299
--- /dev/null
+++ b/pkg/tern/local_resume_engine_logging_integration_test.go
@@ -0,0 +1,106 @@
+//go:build integration
+
+package tern
+
+import (
+	"database/sql"
+	"io"
+	"log/slog"
+	"testing"
+	"time"
+
+	"github.com/block/spirit/pkg/utils"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	ternv1 "github.com/block/schemabot/pkg/proto/ternv1"
+	"github.com/block/schemabot/pkg/state"
+	"github.com/block/schemabot/pkg/storage"
+)
+
+// An apply changes hands whenever a driver dies mid-drive and a peer reclaims
+// it, and the drive that finishes the work is a resumed one. The engine's own
+// log lines have to keep reaching the apply log stream across that handover:
+// that stream is what the CLI and the PR summary render, so an apply resumed by
+// a second driver would otherwise go quiet for the whole drive that actually
+// ran it.
+func TestLocalClient_ResumedDriveCapturesEngineLogs(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	_, dsn := setupMySQLContainer(t)
+	setupStorageSchema(t, dsn)
+	cleanupTasks(t, dsn)
+	cleanupTestTables(t, dsn)
+
+	ctx := t.Context()
+	db, err := sql.Open("mysql", dsn)
+	require.NoError(t, err, "open target database")
+	defer utils.CloseAndLog(db)
+	_, err = db.ExecContext(ctx, "CREATE TABLE users (id INT PRIMARY KEY)")
+	require.NoError(t, err, "create target table")
+	_, err = db.ExecContext(ctx, "INSERT INTO users (id) VALUES (1), (2), (3)")
+	require.NoError(t, err, "seed the target table")
+
+	// The engine routes a Spirit line into the apply log stream from its log
+	// handler, so the client's logger has to admit info records for the routing
+	// to run at all — the same level a server runs at. Discard the output; the
+	// assertion reads the stored stream, not stdout.
+	logger := slog.New(slog.NewTextHandler(io.Discard, &slog.HandlerOptions{Level: slog.LevelInfo}))
+	stor := createStorage(t, dsn)
+	client, err := NewLocalClient(LocalConfig{
+		Database:  "testdb",
+		Type:      storage.DatabaseTypeMySQL,
+		TargetDSN: dsn,
+	}, stor, logger)
+	require.NoError(t, err, "create local client")
+	defer utils.CloseAndLog(client)
+
+	// A column-type change rebuilds the table, so the engine runs a real copy
+	// and emits the log lines this test is about.
+	schemaFiles := buildSchemaWithAllTables(t, dsn, map[string]string{
+		"users": "CREATE TABLE users (id BIGINT PRIMARY KEY, email VARCHAR(255))",
+	})
+	planResp, err := client.Plan(ctx, &ternv1.PlanRequest{
+		Type:     "mysql",
+		Database: "testdb",
+		SchemaFiles: map[string]*ternv1.SchemaFiles{
+			"testdb": {Files: schemaFiles},
+		},
+	})
+	require.NoError(t, err, "plan the schema change")
+	applyResp, err := client.Apply(ctx, &ternv1.ApplyRequest{
+		PlanId:      planResp.PlanId,
+		Environment: localClientTestEnvironment,
+	})
+	require.NoError(t, err, "dispatch the apply")
+	require.True(t, applyResp.Accepted, "dispatch rejected: %s", applyResp.ErrorMessage)
+
+	// Put the apply in the shape a drive that failed part-way leaves behind:
+	// started, paused for operator retry, with its table work still to do. The
+	// next claim is a resume, not a first drive.
+	apply := resolveDispatchedApply(t, stor, applyResp.ApplyId)
+	startedAt := time.Now()
+	apply.State = state.Apply.FailedRetryable
+	apply.StartedAt = &startedAt
+	require.NoError(t, stor.Applies().Update(ctx, apply), "pause the apply for operator retry")
+
+	driveQueuedApply(t, stor, client, applyResp.ApplyId)
+
+	settled, err := stor.Applies().Get(ctx, apply.ID)
+	require.NoError(t, err, "reload the resumed apply")
+	require.NotNil(t, settled)
+	require.Equal(t, state.Apply.Completed, settled.State, "the resumed drive must finish the schema change")
+
+	logs, err := stor.ApplyLogs().GetRecentByApply(ctx, apply.ID, 200)
+	require.NoError(t, err, "load the apply's log stream")
+	var engineLines int
+	for _, entry := range logs {
+		if entry.Source == storage.LogSourceSpirit {
+			engineLines++
+		}
+	}
+	assert.Positive(t, engineLines,
+		"a resumed drive must route the engine's log lines into the apply log stream")
+}

From d99f744359a510d773aa09fff1c7abb85fce0698 Mon Sep 17 00:00:00 2001
From: Armand Parajon <armand@squareup.com>
Date: Thu, 13 Aug 2026 18:51:39 +0800
Subject: [PATCH 2/6] feat(observability): record a failed apply's cause in its
 own log stream
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A failing apply wrote nothing to the apply log: the failure and retryable
helpers moved state and left the reason in the server logs alone, so the
CLI and the PR summary — which render that stream — showed an apply going
terminal with nothing stating why. Both helpers now record the transition
they make, and expiry records why operator recovery stopped retrying, so
the stream carries the whole arc from the first paused attempt to the
permanent failure.

The failure record now has a single owner, so the call sites that logged
their own copy no longer do.
---
 pkg/api/operator.go                |  33 ++++++++++
 pkg/api/operator_test.go           |  33 ++++++++++
 pkg/tern/apply_failure_log_test.go | 101 +++++++++++++++++++++++++++++
 pkg/tern/local_apply_failure.go    |  17 +++++
 pkg/tern/local_apply_grouped.go    |  10 +--
 pkg/tern/local_control_resume.go   |   6 +-
 6 files changed, 186 insertions(+), 14 deletions(-)
 create mode 100644 pkg/tern/apply_failure_log_test.go

diff --git a/pkg/api/operator.go b/pkg/api/operator.go
index 01969181b..f946f044b 100644
--- a/pkg/api/operator.go
+++ b/pkg/api/operator.go
@@ -238,6 +238,39 @@ func (s *Service) expireRetryableApplies(ctx context.Context, driverID int) {
 				"attempt", apply.Attempt,
 				"reason", expiration.Reason)...)
 		metrics.RecordOperatorResumeFailure(ctx, apply.Database, apply.Deployment, apply.Environment, string(expiration.Reason))
+		s.logApplyExpiration(ctx, apply, expiration.Reason)
+	}
+}
+
+// logApplyExpiration appends a durable apply log entry recording that operator
+// recovery gave up on the apply. Expiry is what makes a retryable failure
+// permanent, so without it the apply log ends on the last paused attempt and an
+// operator reading the CLI or the PR summary sees the apply reach a terminal
+// state with nothing stating why. Best-effort: a failed append must not stop
+// the driver from expiring the remaining applies.
+func (s *Service) logApplyExpiration(ctx context.Context, apply *storage.Apply, reason storage.RetryableExpirationReason) {
+	logStore := s.storage.ApplyLogs()
+	if logStore == nil {
+		s.logger.Warn("operator: no apply log store configured; apply expiry will not appear in apply logs",
+			apply.LogAttrs()...)
+		return
+	}
+	message := fmt.Sprintf("Operator recovery gave up on the apply after %d of %d attempts (%s); it will not be retried automatically",
+		apply.Attempt, storage.MaxRecoveryAttempts, reason)
+	logCtx, cancel := context.WithTimeout(ctx, ApplyClaimLogTimeout)
+	defer cancel()
+	if err := logStore.Append(logCtx, &storage.ApplyLog{
+		ApplyID:   apply.ID,
+		Level:     storage.LogLevelError,
+		EventType: storage.LogEventError,
+		Source:    storage.LogSourceSchemaBot,
+		Message:   message,
+		OldState:  state.Apply.FailedRetryable,
+		NewState:  state.Apply.Failed,
+		CreatedAt: s.clock.Now(),
+	}); err != nil {
+		s.logger.Warn("operator: failed to log apply expiry; the apply's own log will not state why recovery stopped",
+			append(apply.LogAttrs(), "error", err)...)
 	}
 }
 
diff --git a/pkg/api/operator_test.go b/pkg/api/operator_test.go
index b9ee2c594..dc9438435 100644
--- a/pkg/api/operator_test.go
+++ b/pkg/api/operator_test.go
@@ -242,6 +242,39 @@ func (s *expiringApplyStore) ExpireRetryable(context.Context) ([]*storage.Retrya
 	return s.expirations, nil
 }
 
+// Expiry is what makes a retryable failure permanent, so it belongs in the
+// apply's own log stream: that stream is what the CLI and the PR summary
+// render, and an apply whose last entry is a paused attempt reads as one that
+// went terminal for no stated reason.
+func TestExpireRetryableApplies_RecordsWhyRecoveryStoppedInTheApplyLog(t *testing.T) {
+	apply := &storage.Apply{
+		ID:              42,
+		ApplyIdentifier: "apply-42",
+		Database:        "appdb",
+		Environment:     "staging",
+		State:           state.Apply.Failed,
+		Attempt:         storage.MaxRecoveryAttempts,
+	}
+	applyLogs := &capturingApplyLogStore{}
+	svc := New(&mockStorageWithApplyStores{
+		applies: &expiringApplyStore{expirations: []*storage.RetryableApplyExpiration{
+			{Apply: apply, Reason: storage.RetryableExpirationAttemptBudget},
+		}},
+		applyLogs: applyLogs,
+	}, testServerConfig(), nil, slog.Default())
+
+	svc.expireRetryableApplies(t.Context(), 1)
+
+	require.Len(t, applyLogs.logs, 1)
+	entry := applyLogs.logs[0]
+	assert.Equal(t, storage.LogLevelError, entry.Level)
+	assert.Equal(t, int64(42), entry.ApplyID)
+	assert.Contains(t, entry.Message, "10 of 10 attempts")
+	assert.Contains(t, entry.Message, string(storage.RetryableExpirationAttemptBudget))
+	assert.Equal(t, state.Apply.FailedRetryable, entry.OldState)
+	assert.Equal(t, state.Apply.Failed, entry.NewState)
+}
+
 // A retryable-apply expiry is a control-plane lifecycle transition an operator
 // triages from logs alone, so the expiry line must carry the apply's full
 // triage attributes — including external_id, the join key to the data plane's
diff --git a/pkg/tern/apply_failure_log_test.go b/pkg/tern/apply_failure_log_test.go
new file mode 100644
index 000000000..8f613dff2
--- /dev/null
+++ b/pkg/tern/apply_failure_log_test.go
@@ -0,0 +1,101 @@
+package tern
+
+import (
+	"context"
+	"log/slog"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/block/schemabot/pkg/state"
+	"github.com/block/schemabot/pkg/storage"
+)
+
+// capturingApplyLogStore records every entry written to an apply's log stream.
+type capturingApplyLogStore struct {
+	storage.ApplyLogStore
+	entries []*storage.ApplyLog
+}
+
+func (s *capturingApplyLogStore) Append(_ context.Context, entry *storage.ApplyLog) error {
+	s.entries = append(s.entries, entry)
+	return nil
+}
+
+func newFailureLogTestClient(apply *storage.Apply, tasks []*storage.Task) (*LocalClient, *capturingApplyLogStore) {
+	logs := &capturingApplyLogStore{}
+	return &LocalClient{
+		config: LocalConfig{Database: "testdb", Type: storage.DatabaseTypeMySQL},
+		storage: &controlTestStorage{
+			applies:         &controlTestApplyStore{apply: apply},
+			tasks:           &controlTestTaskStore{tasks: tasks},
+			applyLogs:       logs,
+			controlRequests: &testControlRequestStore{},
+		},
+		logger: slog.Default(),
+	}, logs
+}
+
+func failureLogTestApply(applyState string, attempt int) *storage.Apply {
+	return &storage.Apply{
+		ID:              7,
+		ApplyIdentifier: "apply-7",
+		Database:        "orders",
+		Environment:     "staging",
+		State:           applyState,
+		Attempt:         attempt,
+	}
+}
+
+// The apply log is the only failure record an operator reads from the CLI or
+// the PR summary comment, so an apply that fails must say so there — not only
+// in the server logs, where a reader of the apply's own history would see it
+// reach a terminal state with nothing stating why.
+func TestFailApplyWithTasksRecordsTheFailureInTheApplyLog(t *testing.T) {
+	apply := failureLogTestApply(state.Apply.Running, 0)
+	task := &storage.Task{ID: 1, TaskIdentifier: "task-1", ApplyID: apply.ID, TableName: "orders", State: state.Task.Running}
+	client, logs := newFailureLogTestClient(apply, []*storage.Task{task})
+
+	client.failApplyWithTasks(t.Context(), apply, []*storage.Task{task}, "engine lost its connection to the target")
+
+	require.Len(t, logs.entries, 1)
+	entry := logs.entries[0]
+	assert.Equal(t, storage.LogLevelError, entry.Level)
+	assert.Equal(t, storage.LogEventError, entry.EventType)
+	assert.Equal(t, "Apply failed: engine lost its connection to the target", entry.Message)
+	assert.Equal(t, state.Apply.Running, entry.OldState)
+	assert.Equal(t, state.Apply.Failed, entry.NewState)
+}
+
+// A retryable failure is the state operator recovery keeps re-driving, so each
+// paused attempt records the budget it spent. Without it the apply log shows
+// only the gaps between attempts, and the retry budget drains invisibly.
+func TestMarkApplyRetryableWithTasksRecordsTheSpentAttempt(t *testing.T) {
+	apply := failureLogTestApply(state.Apply.Running, 3)
+	task := &storage.Task{ID: 1, TaskIdentifier: "task-1", ApplyID: apply.ID, TableName: "orders", State: state.Task.Running}
+	client, logs := newFailureLogTestClient(apply, []*storage.Task{task})
+
+	client.markApplyRetryableWithTasks(t.Context(), apply, []*storage.Task{task}, "target refused the connection")
+
+	require.Len(t, logs.entries, 1)
+	entry := logs.entries[0]
+	assert.Equal(t, storage.LogLevelWarn, entry.Level, "a paused attempt is not yet a permanent failure")
+	assert.Contains(t, entry.Message, "attempt 4 of 10")
+	assert.Contains(t, entry.Message, "target refused the connection")
+	assert.Equal(t, state.Apply.Running, entry.OldState)
+	assert.Equal(t, state.Apply.FailedRetryable, entry.NewState)
+}
+
+// An apply that another driver already settled is not this drive's to fail: the
+// stored verdict stands, and writing a second failure record would report a
+// state transition that never happened.
+func TestFailApplyWithTasksLeavesASettledApplyUnrecorded(t *testing.T) {
+	apply := failureLogTestApply(state.Apply.Cancelled, 0)
+	client, logs := newFailureLogTestClient(apply, nil)
+
+	client.failApplyWithTasks(t.Context(), apply, nil, "engine lost its connection to the target")
+
+	assert.Empty(t, logs.entries)
+	assert.Equal(t, state.Apply.Cancelled, apply.State)
+}
diff --git a/pkg/tern/local_apply_failure.go b/pkg/tern/local_apply_failure.go
index 2dd880b8e..fc1e86e55 100644
--- a/pkg/tern/local_apply_failure.go
+++ b/pkg/tern/local_apply_failure.go
@@ -2,6 +2,7 @@ package tern
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/block/schemabot/pkg/metrics"
@@ -38,12 +39,20 @@ func (c *LocalClient) failApplyWithTasks(ctx context.Context, apply *storage.App
 		return
 	}
 
+	previousState := apply.State
 	apply.State = state.Apply.Failed
 	apply.ErrorMessage = errMsg
 	apply.CompletedAt = &now
 	apply.UpdatedAt = now
 	if err := c.storage.Applies().Update(ctx, apply); err != nil {
 		logger.Error("failed to update apply state", append(apply.MutableLogAttrs(), "error", err)...)
+	} else {
+		// Record the failure in the apply's own log stream. It is the only
+		// surface an operator reads from the CLI or the PR summary, so a failure
+		// that lands only in the server logs reads there as an apply that went
+		// terminal for no stated reason.
+		c.logApplyEvent(ctx, apply.ID, nil, storage.LogLevelError, storage.LogEventError, storage.LogSourceSchemaBot,
+			fmt.Sprintf("Apply failed: %s", errMsg), previousState, state.Apply.Failed)
 	}
 	metrics.AdjustActiveApplies(ctx, -1, apply.Database, apply.Deployment, apply.Environment)
 }
@@ -74,12 +83,20 @@ func (c *LocalClient) markApplyRetryableWithTasks(ctx context.Context, apply *st
 		return
 	}
 
+	previousState := apply.State
 	apply.State = state.Apply.FailedRetryable
 	apply.ErrorMessage = errMsg
 	apply.CompletedAt = nil
 	apply.UpdatedAt = time.Now()
 	if err := c.storage.Applies().Update(ctx, apply); err != nil {
 		logger.Error("failed to update apply state", append(apply.MutableLogAttrs(), "error", err)...)
+	} else {
+		// Each paused attempt is recorded with the budget it spent, so the apply
+		// log shows recovery burning through its attempts rather than only the
+		// silence between them.
+		c.logApplyEvent(ctx, apply.ID, nil, storage.LogLevelWarn, storage.LogEventError, storage.LogSourceSchemaBot,
+			fmt.Sprintf("Apply paused for operator retry (attempt %d of %d): %s", apply.Attempt+1, storage.MaxRecoveryAttempts, errMsg),
+			previousState, state.Apply.FailedRetryable)
 	}
 	metrics.AdjustActiveApplies(ctx, -1, apply.Database, apply.Deployment, apply.Environment)
 	if obs := c.getObserver(apply.ID); obs != nil {
diff --git a/pkg/tern/local_apply_grouped.go b/pkg/tern/local_apply_grouped.go
index dd5c74b92..088a0e2dc 100644
--- a/pkg/tern/local_apply_grouped.go
+++ b/pkg/tern/local_apply_grouped.go
@@ -130,19 +130,11 @@ func (c *LocalClient) executeGroupedApply(ctx context.Context, apply *storage.Ap
 		} else {
 			logger.Error("apply failed", append(apply.MutableLogAttrs(), "mode", mode, "error", err)...)
 		}
-		logLevel := storage.LogLevelError
-		if newState == state.Apply.FailedRetryable {
-			logLevel = storage.LogLevelWarn
-		}
-		c.logApplyEvent(ctx, apply.ID, nil, logLevel, storage.LogEventError, storage.LogSourceSchemaBot,
-			fmt.Sprintf("Engine apply failed: %v", err), state.Apply.Pending, newState)
 		return
 	}
 
 	if !result.Accepted {
-		c.failApplyWithTasks(ctx, apply, tasks, result.Message)
-		c.logApplyEvent(ctx, apply.ID, nil, storage.LogLevelError, storage.LogEventError, storage.LogSourceSchemaBot,
-			fmt.Sprintf("Engine apply not accepted: %s", result.Message), state.Apply.Pending, state.Apply.Failed)
+		c.failApplyWithTasks(ctx, apply, tasks, fmt.Sprintf("engine did not accept the apply: %s", result.Message))
 		return
 	}
 
diff --git a/pkg/tern/local_control_resume.go b/pkg/tern/local_control_resume.go
index 20280b017..15d468464 100644
--- a/pkg/tern/local_control_resume.go
+++ b/pkg/tern/local_control_resume.go
@@ -1591,8 +1591,6 @@ func (c *LocalClient) resumeApplyWithTasks(ctx context.Context, apply *storage.A
 		message := "deferred cutover signal is absent but live schema does not match desired schema; manual reconciliation required"
 		logger.Error("deferred cutover recovery cannot reconcile absent cutover signal",
 			"active_task_count", len(activeTasks))
-		c.logApplyEvent(ctx, apply.ID, nil, storage.LogLevelError, storage.LogEventError, storage.LogSourceSchemaBot,
-			message, apply.State, state.Apply.Failed)
 		c.failApplyWithTasks(ctx, apply, activeTasks, message)
 		c.notifyTerminalObserver(apply, tasks)
 		return nil
@@ -1676,9 +1674,7 @@ func (c *LocalClient) handleGroupedResumeFailure(ctx context.Context, apply *sto
 
 	logger.Error("engine apply failed during recovery",
 		"error", err)
-	c.logApplyEvent(ctx, apply.ID, nil, storage.LogLevelError, storage.LogEventError, storage.LogSourceSchemaBot,
-		fmt.Sprintf("Recovery failed: %v", err), apply.State, state.Apply.Failed)
-	c.failApplyWithTasks(ctx, apply, tasks, err.Error())
+	c.failApplyWithTasks(ctx, apply, tasks, fmt.Sprintf("recovery failed: %v", err))
 	if startRequested {
 		if failErr := failPendingControlRequests(ctx, c.storage, apply, storage.ControlOperationStart, err.Error()); failErr != nil {
 			return failErr

From 84ed2c3737cbc6a94edbc48c0e846f12074cb363 Mon Sep 17 00:00:00 2001
From: Armand Parajon <armand@squareup.com>
Date: Fri, 14 Aug 2026 16:46:19 +0800
Subject: [PATCH 3/6] fix(observability): keep a detached resume's log capture
 alive and its attempt count in budget
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A detached grouped resume polls under a context that deliberately outlives
its caller, but its engine log wiring held the caller's context — so capture
ended the moment the caller returned and the rest of the schema change went
unrecorded. The wiring is now bound to the context the poll runs under.

The paused-attempt record counted drives against the recovery budget, two
quantities that diverge because a recovery claim advances the apply's attempt
counter. At the last attempt the record read past its own limit. It now
reports the attempt counter the budget is measured against.
---
 pkg/tern/apply_failure_log_test.go | 21 +++++++++++++++++++--
 pkg/tern/local_apply_failure.go    |  6 ++++--
 pkg/tern/local_control_resume.go   |  5 +++++
 3 files changed, 28 insertions(+), 4 deletions(-)

diff --git a/pkg/tern/apply_failure_log_test.go b/pkg/tern/apply_failure_log_test.go
index 8f613dff2..1917a333f 100644
--- a/pkg/tern/apply_failure_log_test.go
+++ b/pkg/tern/apply_failure_log_test.go
@@ -2,6 +2,7 @@ package tern
 
 import (
 	"context"
+	"fmt"
 	"log/slog"
 	"testing"
 
@@ -70,7 +71,9 @@ func TestFailApplyWithTasksRecordsTheFailureInTheApplyLog(t *testing.T) {
 
 // A retryable failure is the state operator recovery keeps re-driving, so each
 // paused attempt records the budget it spent. Without it the apply log shows
-// only the gaps between attempts, and the retry budget drains invisibly.
+// only the gaps between attempts, and the retry budget drains invisibly. The
+// count reported is the apply's own attempt counter, so it can never exceed the
+// budget it is measured against.
 func TestMarkApplyRetryableWithTasksRecordsTheSpentAttempt(t *testing.T) {
 	apply := failureLogTestApply(state.Apply.Running, 3)
 	task := &storage.Task{ID: 1, TaskIdentifier: "task-1", ApplyID: apply.ID, TableName: "orders", State: state.Task.Running}
@@ -81,12 +84,26 @@ func TestMarkApplyRetryableWithTasksRecordsTheSpentAttempt(t *testing.T) {
 	require.Len(t, logs.entries, 1)
 	entry := logs.entries[0]
 	assert.Equal(t, storage.LogLevelWarn, entry.Level, "a paused attempt is not yet a permanent failure")
-	assert.Contains(t, entry.Message, "attempt 4 of 10")
+	assert.Contains(t, entry.Message, "3 of 10 recovery attempts used")
 	assert.Contains(t, entry.Message, "target refused the connection")
 	assert.Equal(t, state.Apply.Running, entry.OldState)
 	assert.Equal(t, state.Apply.FailedRetryable, entry.NewState)
 }
 
+// The last drive operator recovery is allowed to make still reports a count
+// inside the budget, so an operator reading the apply log sees the retry budget
+// reach its limit rather than a figure that exceeds it.
+func TestMarkApplyRetryableWithTasksReportsTheFinalAttemptWithinBudget(t *testing.T) {
+	apply := failureLogTestApply(state.Apply.Running, storage.MaxRecoveryAttempts)
+	client, logs := newFailureLogTestClient(apply, nil)
+
+	client.markApplyRetryableWithTasks(t.Context(), apply, nil, "target refused the connection")
+
+	require.Len(t, logs.entries, 1)
+	assert.Contains(t, logs.entries[0].Message,
+		fmt.Sprintf("%d of %d recovery attempts used", storage.MaxRecoveryAttempts, storage.MaxRecoveryAttempts))
+}
+
 // An apply that another driver already settled is not this drive's to fail: the
 // stored verdict stands, and writing a second failure record would report a
 // state transition that never happened.
diff --git a/pkg/tern/local_apply_failure.go b/pkg/tern/local_apply_failure.go
index fc1e86e55..a9020e6a8 100644
--- a/pkg/tern/local_apply_failure.go
+++ b/pkg/tern/local_apply_failure.go
@@ -93,9 +93,11 @@ func (c *LocalClient) markApplyRetryableWithTasks(ctx context.Context, apply *st
 	} else {
 		// Each paused attempt is recorded with the budget it spent, so the apply
 		// log shows recovery burning through its attempts rather than only the
-		// silence between them.
+		// silence between them. The count is the apply's own attempt counter,
+		// which a recovery claim advances — reporting the drive number instead
+		// would run past the budget it is measured against.
 		c.logApplyEvent(ctx, apply.ID, nil, storage.LogLevelWarn, storage.LogEventError, storage.LogSourceSchemaBot,
-			fmt.Sprintf("Apply paused for operator retry (attempt %d of %d): %s", apply.Attempt+1, storage.MaxRecoveryAttempts, errMsg),
+			fmt.Sprintf("Apply paused for operator retry (%d of %d recovery attempts used): %s", apply.Attempt, storage.MaxRecoveryAttempts, errMsg),
 			previousState, state.Apply.FailedRetryable)
 	}
 	metrics.AdjustActiveApplies(ctx, -1, apply.Database, apply.Deployment, apply.Environment)
diff --git a/pkg/tern/local_control_resume.go b/pkg/tern/local_control_resume.go
index 15d468464..cfce73846 100644
--- a/pkg/tern/local_control_resume.go
+++ b/pkg/tern/local_control_resume.go
@@ -855,6 +855,11 @@ func (c *LocalClient) launchAtomicResume(ctx context.Context, apply *storage.App
 	resumeCtx, cancelResume := context.WithCancel(context.WithoutCancel(ctx))
 	stopHeartbeat := c.startParentApplyHeartbeat(resumeCtx, apply, suppressParent, cancelResume)
 	pollDetached = true
+	// The detached poll deliberately outlives the caller's context, so its log
+	// wiring has to as well: a callback holding the caller's context records
+	// nothing once that context is cancelled, and the engine lines for the rest
+	// of the schema change are exactly the ones worth keeping.
+	stopEngineLogging = c.setupSpiritLogging(resumeCtx, apply, tasks)
 	go func() {
 		defer cancelResume()
 		defer stopHeartbeat()

From 339ed5652457febf674bbea1d9f82cca825c9a43 Mon Sep 17 00:00:00 2001
From: Armand Parajon <armand@squareup.com>
Date: Fri, 14 Aug 2026 17:26:01 +0800
Subject: [PATCH 4/6] fix(observability): keep the engine's own words as a
 failed apply's stored cause
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Consolidating the failure record onto one owner moved two call sites' framing
into the message passed to failApplyWithTasks, which is stored as the apply's
and every task's error message — not just written to the log. That prefixed a
cause an operator reads in status output and PR comments with wording meant
for the log line.

The stored cause is the engine's text again. The apply log entry carries its
own framing, and the recovery entry that precedes it already establishes that
the drive was a resumed one.
---
 pkg/tern/local_apply_grouped.go           | 2 +-
 pkg/tern/local_client_integration_test.go | 2 +-
 pkg/tern/local_control_resume.go          | 2 +-
 3 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/pkg/tern/local_apply_grouped.go b/pkg/tern/local_apply_grouped.go
index 088a0e2dc..7d6ca1f80 100644
--- a/pkg/tern/local_apply_grouped.go
+++ b/pkg/tern/local_apply_grouped.go
@@ -134,7 +134,7 @@ func (c *LocalClient) executeGroupedApply(ctx context.Context, apply *storage.Ap
 	}
 
 	if !result.Accepted {
-		c.failApplyWithTasks(ctx, apply, tasks, fmt.Sprintf("engine did not accept the apply: %s", result.Message))
+		c.failApplyWithTasks(ctx, apply, tasks, result.Message)
 		return
 	}
 
diff --git a/pkg/tern/local_client_integration_test.go b/pkg/tern/local_client_integration_test.go
index 592576481..9e5c511b9 100644
--- a/pkg/tern/local_client_integration_test.go
+++ b/pkg/tern/local_client_integration_test.go
@@ -2920,7 +2920,7 @@ func TestLocalClient_ResumeApplyGroupedStartRequestFailsWhenEngineRejects(t *tes
 
 	logs, err := stor.ApplyLogs().GetByApply(ctx, applyID)
 	require.NoError(t, err)
-	assert.True(t, hasLogMessageContaining(logs, "Recovery failed: engine apply failed: engine refused grouped resume"))
+	assert.True(t, hasLogMessageContaining(logs, "Apply failed: engine apply failed: engine refused grouped resume"))
 }
 
 // This scenario covers restart recovery of a grouped Vitess apply whose opaque
diff --git a/pkg/tern/local_control_resume.go b/pkg/tern/local_control_resume.go
index cfce73846..229e9ddf8 100644
--- a/pkg/tern/local_control_resume.go
+++ b/pkg/tern/local_control_resume.go
@@ -1679,7 +1679,7 @@ func (c *LocalClient) handleGroupedResumeFailure(ctx context.Context, apply *sto
 
 	logger.Error("engine apply failed during recovery",
 		"error", err)
-	c.failApplyWithTasks(ctx, apply, tasks, fmt.Sprintf("recovery failed: %v", err))
+	c.failApplyWithTasks(ctx, apply, tasks, err.Error())
 	if startRequested {
 		if failErr := failPendingControlRequests(ctx, c.storage, apply, storage.ControlOperationStart, err.Error()); failErr != nil {
 			return failErr

From d105288e6bd1e2cb4ab967088aa8d11513bb1684 Mon Sep 17 00:00:00 2001
From: Armand Parajon <armand@squareup.com>
Date: Sat, 15 Aug 2026 09:34:57 +0800
Subject: [PATCH 5/6] fix(observability): record a sequential apply's failure
 in its own log
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The sequential path is the one most applies take, and it wrote the
failure only to the server logs — so the apply's own history showed it
reach a terminal state with nothing stating why, on exactly the path an
operator is most likely to be reading. Route it through the same log
owners as every other path, and report the retry budget as attempts
remaining so the first pause reads as a countdown rather than "0 of 10".

Also folds the operator's three near-identical best-effort apply-log
appends into one helper, so a future contract change cannot miss one.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 docs/apply-lifecycle.md            | 11 +++--
 pkg/api/operator.go                | 70 +++++++++++----------------
 pkg/api/operator_test.go           |  6 ++-
 pkg/tern/apply_failure_log_test.go | 77 ++++++++++++++++++++++++++----
 pkg/tern/local_apply_failure.go    | 38 +++++++++------
 pkg/tern/local_apply_sequential.go | 11 +++++
 6 files changed, 140 insertions(+), 73 deletions(-)

diff --git a/docs/apply-lifecycle.md b/docs/apply-lifecycle.md
index 6c4e2bc4f..692b7df76 100644
--- a/docs/apply-lifecycle.md
+++ b/docs/apply-lifecycle.md
@@ -221,10 +221,13 @@ budget and window are per-apply — a new apply starts with a full budget.
 1. **Confirm the state.** Check the apply's status (CLI status command, or the
    summary comment on the PR). `failed_retryable` needs no action — recovery
    retries it automatically. Only permanent `failed` needs you.
-2. **Read the failure.** The apply's error message and the server logs (keyed
-   by the apply identifier) say what the engine rejected or what broke.
-   Fix the underlying cause if there is one — an invalid statement, a
-   permissions problem, an unreachable target.
+2. **Read the failure.** The apply's own log stream carries the cause, each
+   paused attempt with the recovery budget it had left, and the reason recovery
+   gave up — read it from the CLI (`schemabot logs <apply>`) or the PR summary
+   comment, no infrastructure access required. The server logs, keyed by the
+   apply identifier, are there when you need what happened around the apply
+   rather than to it. Fix the underlying cause if there is one — an invalid
+   statement, a permissions problem, an unreachable target.
 3. **Plan again.** Run a fresh plan against the same schema and environment
    (`schemabot plan`, or a new `schemabot plan` PR comment). The plan reflects
    whatever the failed apply already completed: tables that landed are not in
diff --git a/pkg/api/operator.go b/pkg/api/operator.go
index f946f044b..bc7bb3530 100644
--- a/pkg/api/operator.go
+++ b/pkg/api/operator.go
@@ -249,28 +249,36 @@ func (s *Service) expireRetryableApplies(ctx context.Context, driverID int) {
 // state with nothing stating why. Best-effort: a failed append must not stop
 // the driver from expiring the remaining applies.
 func (s *Service) logApplyExpiration(ctx context.Context, apply *storage.Apply, reason storage.RetryableExpirationReason) {
-	logStore := s.storage.ApplyLogs()
-	if logStore == nil {
-		s.logger.Warn("operator: no apply log store configured; apply expiry will not appear in apply logs",
-			apply.LogAttrs()...)
-		return
-	}
-	message := fmt.Sprintf("Operator recovery gave up on the apply after %d of %d attempts (%s); it will not be retried automatically",
-		apply.Attempt, storage.MaxRecoveryAttempts, reason)
-	logCtx, cancel := context.WithTimeout(ctx, ApplyClaimLogTimeout)
-	defer cancel()
-	if err := logStore.Append(logCtx, &storage.ApplyLog{
+	s.appendApplyLog(ctx, s.logger, &storage.ApplyLog{
 		ApplyID:   apply.ID,
 		Level:     storage.LogLevelError,
 		EventType: storage.LogEventError,
 		Source:    storage.LogSourceSchemaBot,
-		Message:   message,
+		Message: fmt.Sprintf("Operator recovery gave up on the apply after %d of %d attempts (%s); it will not be retried automatically",
+			apply.Attempt, storage.MaxRecoveryAttempts, reason),
 		OldState:  state.Apply.FailedRetryable,
 		NewState:  state.Apply.Failed,
 		CreatedAt: s.clock.Now(),
-	}); err != nil {
-		s.logger.Warn("operator: failed to log apply expiry; the apply's own log will not state why recovery stopped",
-			append(apply.LogAttrs(), "error", err)...)
+	}, "why recovery stopped", apply.LogAttrs()...)
+}
+
+// appendApplyLog writes one entry to the apply's own log stream, bounded so a
+// slow store cannot stall the driver. It is best-effort by contract: every
+// caller has already done the work the entry describes, and an entry that
+// cannot be written must not undo it. record names what an operator loses when
+// the entry does not land, so the warning says which part of the apply's
+// account is missing rather than only that a write failed.
+func (s *Service) appendApplyLog(ctx context.Context, logger *slog.Logger, entry *storage.ApplyLog, record string, logAttrs ...any) {
+	logStore := s.storage.ApplyLogs()
+	if logStore == nil {
+		logger.Warn("operator: no apply log store configured; the apply's own log will not state "+record, logAttrs...)
+		return
+	}
+	logCtx, cancel := context.WithTimeout(ctx, ApplyClaimLogTimeout)
+	defer cancel()
+	if err := logStore.Append(logCtx, entry); err != nil {
+		logger.Warn("operator: failed to append to the apply log; the apply's own log will not state "+record,
+			append(slices.Clone(logAttrs), "error", err)...)
 	}
 }
 
@@ -1449,14 +1457,7 @@ func (s *Service) resumeClaimedApplyWithOptions(ctx context.Context, driverID in
 // append must not block the resume, so the error is logged on the caller's
 // drive-scoped logger and the claim proceeds.
 func (s *Service) logApplyResumeClaim(ctx context.Context, logger *slog.Logger, driverID int, apply *storage.Apply) {
-	logStore := s.storage.ApplyLogs()
-	if logStore == nil {
-		logger.Warn("operator: no apply log store configured; apply claim will not appear in apply logs")
-		return
-	}
-	logCtx, cancel := context.WithTimeout(ctx, ApplyClaimLogTimeout)
-	defer cancel()
-	if err := logStore.Append(logCtx, &storage.ApplyLog{
+	s.appendApplyLog(ctx, logger, &storage.ApplyLog{
 		ApplyID:   apply.ID,
 		Level:     storage.LogLevelInfo,
 		EventType: storage.LogEventInfo,
@@ -1465,10 +1466,7 @@ func (s *Service) logApplyResumeClaim(ctx context.Context, logger *slog.Logger,
 		OldState:  apply.State,
 		NewState:  apply.State,
 		CreatedAt: s.clock.Now(),
-	}); err != nil {
-		logger.Warn("operator: failed to log apply claim; apply claim will not appear in apply logs",
-			"error", err)
-	}
+	}, "that a driver claimed it to resume it")
 }
 
 // failClaimedApplyAfterDrivePanic contains an engine-drive panic to the
@@ -1581,16 +1579,7 @@ func (s *Service) failClaimedApplyAfterDrivePanic(ctx context.Context, driverID
 // state without server logs. Best-effort: a failed append must not block
 // containment.
 func (s *Service) logApplyDrivePanicFailure(ctx context.Context, driverID int, apply *storage.Apply, previousState, errMsg string) {
-	logStore := s.storage.ApplyLogs()
-	if logStore == nil {
-		s.logger.Warn("operator: no apply log store configured; the drive panic will not appear in apply logs",
-			append(apply.LogAttrs(),
-				"driver", driverID)...)
-		return
-	}
-	logCtx, cancel := context.WithTimeout(ctx, ApplyClaimLogTimeout)
-	defer cancel()
-	if err := logStore.Append(logCtx, &storage.ApplyLog{
+	s.appendApplyLog(ctx, s.logger, &storage.ApplyLog{
 		ApplyID:   apply.ID,
 		Level:     storage.LogLevelError,
 		EventType: storage.LogEventError,
@@ -1599,12 +1588,7 @@ func (s *Service) logApplyDrivePanicFailure(ctx context.Context, driverID int, a
 		OldState:  previousState,
 		NewState:  apply.State,
 		CreatedAt: s.clock.Now(),
-	}); err != nil {
-		s.logger.Warn("operator: failed to log drive panic failure; the failure will not appear in apply logs",
-			append(apply.LogAttrs(),
-				"driver", driverID,
-				"error", err)...)
-	}
+	}, "that a contained drive panic failed it", append(apply.LogAttrs(), "driver", driverID)...)
 }
 
 // startApplyOperationHeartbeat refreshes the claimed operation row's lease while
diff --git a/pkg/api/operator_test.go b/pkg/api/operator_test.go
index dc9438435..f2683576b 100644
--- a/pkg/api/operator_test.go
+++ b/pkg/api/operator_test.go
@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"log/slog"
 	"os"
 	"sync"
@@ -221,7 +222,7 @@ func TestResumeClaimedApply_DriveLogsCarryApplyIdentity(t *testing.T) {
 
 	// The call site passes only the message — every identity attr must come
 	// from the bound logger.
-	noStore := requireLogLine(t, lines, "operator: no apply log store configured; apply claim will not appear in apply logs")
+	noStore := requireLogLine(t, lines, "operator: no apply log store configured; the apply's own log will not state that a driver claimed it to resume it")
 	assertDriveIdentity(noStore)
 
 	resumedLine := requireLogLine(t, lines, "operator: resumed apply")
@@ -269,7 +270,8 @@ func TestExpireRetryableApplies_RecordsWhyRecoveryStoppedInTheApplyLog(t *testin
 	entry := applyLogs.logs[0]
 	assert.Equal(t, storage.LogLevelError, entry.Level)
 	assert.Equal(t, int64(42), entry.ApplyID)
-	assert.Contains(t, entry.Message, "10 of 10 attempts")
+	assert.Contains(t, entry.Message,
+		fmt.Sprintf("%d of %d attempts", storage.MaxRecoveryAttempts, storage.MaxRecoveryAttempts))
 	assert.Contains(t, entry.Message, string(storage.RetryableExpirationAttemptBudget))
 	assert.Equal(t, state.Apply.FailedRetryable, entry.OldState)
 	assert.Equal(t, state.Apply.Failed, entry.NewState)
diff --git a/pkg/tern/apply_failure_log_test.go b/pkg/tern/apply_failure_log_test.go
index 1917a333f..c4889fbdc 100644
--- a/pkg/tern/apply_failure_log_test.go
+++ b/pkg/tern/apply_failure_log_test.go
@@ -70,11 +70,11 @@ func TestFailApplyWithTasksRecordsTheFailureInTheApplyLog(t *testing.T) {
 }
 
 // A retryable failure is the state operator recovery keeps re-driving, so each
-// paused attempt records the budget it spent. Without it the apply log shows
+// paused attempt records the budget it has left. Without it the apply log shows
 // only the gaps between attempts, and the retry budget drains invisibly. The
-// count reported is the apply's own attempt counter, so it can never exceed the
-// budget it is measured against.
-func TestMarkApplyRetryableWithTasksRecordsTheSpentAttempt(t *testing.T) {
+// countdown is derived from the apply's own attempt counter, so it can never
+// run past the budget it is measured against.
+func TestMarkApplyRetryableWithTasksRecordsTheRemainingBudget(t *testing.T) {
 	apply := failureLogTestApply(state.Apply.Running, 3)
 	task := &storage.Task{ID: 1, TaskIdentifier: "task-1", ApplyID: apply.ID, TableName: "orders", State: state.Task.Running}
 	client, logs := newFailureLogTestClient(apply, []*storage.Task{task})
@@ -84,16 +84,17 @@ func TestMarkApplyRetryableWithTasksRecordsTheSpentAttempt(t *testing.T) {
 	require.Len(t, logs.entries, 1)
 	entry := logs.entries[0]
 	assert.Equal(t, storage.LogLevelWarn, entry.Level, "a paused attempt is not yet a permanent failure")
-	assert.Contains(t, entry.Message, "3 of 10 recovery attempts used")
+	assert.Contains(t, entry.Message,
+		fmt.Sprintf("%d of %d recovery attempts remaining", storage.MaxRecoveryAttempts-3, storage.MaxRecoveryAttempts))
 	assert.Contains(t, entry.Message, "target refused the connection")
 	assert.Equal(t, state.Apply.Running, entry.OldState)
 	assert.Equal(t, state.Apply.FailedRetryable, entry.NewState)
 }
 
-// The last drive operator recovery is allowed to make still reports a count
-// inside the budget, so an operator reading the apply log sees the retry budget
-// reach its limit rather than a figure that exceeds it.
-func TestMarkApplyRetryableWithTasksReportsTheFinalAttemptWithinBudget(t *testing.T) {
+// The last drive operator recovery is allowed to make reports an exhausted
+// budget, so an operator reading the apply log sees the retry budget reach its
+// limit rather than a figure that runs past it.
+func TestMarkApplyRetryableWithTasksReportsAnExhaustedBudget(t *testing.T) {
 	apply := failureLogTestApply(state.Apply.Running, storage.MaxRecoveryAttempts)
 	client, logs := newFailureLogTestClient(apply, nil)
 
@@ -101,7 +102,7 @@ func TestMarkApplyRetryableWithTasksReportsTheFinalAttemptWithinBudget(t *testin
 
 	require.Len(t, logs.entries, 1)
 	assert.Contains(t, logs.entries[0].Message,
-		fmt.Sprintf("%d of %d recovery attempts used", storage.MaxRecoveryAttempts, storage.MaxRecoveryAttempts))
+		fmt.Sprintf("0 of %d recovery attempts remaining", storage.MaxRecoveryAttempts))
 }
 
 // An apply that another driver already settled is not this drive's to fail: the
@@ -116,3 +117,59 @@ func TestFailApplyWithTasksLeavesASettledApplyUnrecorded(t *testing.T) {
 	assert.Empty(t, logs.entries)
 	assert.Equal(t, state.Apply.Cancelled, apply.State)
 }
+
+// The default sequential MySQL path is the one most applies take, so a table
+// that fails there reaches the operator through the apply's own log stream like
+// every other path — with the cause named, and with the budget countdown when
+// the failure is one recovery will re-drive.
+func TestFinalizeSequentialApplyRecordsAPermanentFailure(t *testing.T) {
+	apply := failureLogTestApply(state.Apply.Running, 0)
+	failed := &storage.Task{
+		ID: 1, TaskIdentifier: "task-1", ApplyID: apply.ID, TableName: "orders",
+		State: state.Task.Failed, ErrorMessage: "engine refused the statement",
+	}
+	client, logs := newFailureLogTestClient(apply, []*storage.Task{failed})
+
+	client.finalizeSequentialApply(t.Context(), apply, []*storage.Task{failed}, failed, false)
+
+	require.Len(t, logs.entries, 1)
+	entry := logs.entries[0]
+	assert.Equal(t, storage.LogLevelError, entry.Level)
+	assert.Contains(t, entry.Message, "Apply failed:")
+	assert.Contains(t, entry.Message, "orders")
+	assert.Contains(t, entry.Message, "engine refused the statement")
+	assert.Equal(t, state.Apply.Running, entry.OldState)
+	assert.Equal(t, state.Apply.Failed, entry.NewState)
+}
+
+func TestFinalizeSequentialApplyRecordsARetryablePause(t *testing.T) {
+	apply := failureLogTestApply(state.Apply.Running, 3)
+	failed := &storage.Task{
+		ID: 1, TaskIdentifier: "task-1", ApplyID: apply.ID, TableName: "orders",
+		State: state.Task.FailedRetryable, ErrorMessage: "target refused the connection",
+	}
+	client, logs := newFailureLogTestClient(apply, []*storage.Task{failed})
+
+	client.finalizeSequentialApply(t.Context(), apply, []*storage.Task{failed}, failed, false)
+
+	require.Len(t, logs.entries, 1)
+	entry := logs.entries[0]
+	assert.Equal(t, storage.LogLevelWarn, entry.Level)
+	assert.Contains(t, entry.Message,
+		fmt.Sprintf("%d of %d recovery attempts remaining", storage.MaxRecoveryAttempts-3, storage.MaxRecoveryAttempts))
+	assert.Contains(t, entry.Message, "target refused the connection")
+	assert.Equal(t, state.Apply.FailedRetryable, entry.NewState)
+}
+
+// A sequential apply that succeeds or that an operator stopped is not a failure,
+// and recording one would report a cause that does not exist.
+func TestFinalizeSequentialApplyRecordsNothingWithoutAFailure(t *testing.T) {
+	apply := failureLogTestApply(state.Apply.Running, 0)
+	done := &storage.Task{ID: 1, TaskIdentifier: "task-1", ApplyID: apply.ID, TableName: "orders", State: state.Task.Completed}
+	client, logs := newFailureLogTestClient(apply, []*storage.Task{done})
+
+	client.finalizeSequentialApply(t.Context(), apply, []*storage.Task{done}, nil, false)
+
+	assert.Empty(t, logs.entries)
+	assert.Equal(t, state.Apply.Completed, apply.State)
+}
diff --git a/pkg/tern/local_apply_failure.go b/pkg/tern/local_apply_failure.go
index a9020e6a8..6e40cb313 100644
--- a/pkg/tern/local_apply_failure.go
+++ b/pkg/tern/local_apply_failure.go
@@ -10,6 +10,28 @@ import (
 	"github.com/block/schemabot/pkg/storage"
 )
 
+// logApplyFailure records a permanent failure in the apply's own log stream. It
+// is the only surface an operator reads from the CLI or the PR summary, so a
+// failure that lands only in the server logs reads there as an apply that went
+// terminal for no stated reason. Call it only after the state change is stored,
+// so the log never claims an outcome storage did not take.
+func (c *LocalClient) logApplyFailure(ctx context.Context, apply *storage.Apply, previousState, errMsg string) {
+	c.logApplyEvent(ctx, apply.ID, nil, storage.LogLevelError, storage.LogEventError, storage.LogSourceSchemaBot,
+		fmt.Sprintf("Apply failed: %s", errMsg), previousState, state.Apply.Failed)
+}
+
+// logApplyPausedForRetry records a retryable pause with the budget it has left,
+// so the apply log shows recovery counting down its attempts rather than only
+// the silence between them. The remaining count is derived from the apply's own
+// attempt counter, which a recovery claim advances — counting drives instead
+// would run past the budget it is measured against.
+func (c *LocalClient) logApplyPausedForRetry(ctx context.Context, apply *storage.Apply, previousState, errMsg string) {
+	c.logApplyEvent(ctx, apply.ID, nil, storage.LogLevelWarn, storage.LogEventError, storage.LogSourceSchemaBot,
+		fmt.Sprintf("Apply paused for operator retry (%d of %d recovery attempts remaining): %s",
+			storage.MaxRecoveryAttempts-apply.Attempt, storage.MaxRecoveryAttempts, errMsg),
+		previousState, state.Apply.FailedRetryable)
+}
+
 // failApplyWithTasks marks all tasks and the apply as failed with the given error.
 // If the apply is already in a terminal state (e.g., cancelled by Stop()), the
 // stored state is not overwritten; the settled state is adopted into the
@@ -47,12 +69,7 @@ func (c *LocalClient) failApplyWithTasks(ctx context.Context, apply *storage.App
 	if err := c.storage.Applies().Update(ctx, apply); err != nil {
 		logger.Error("failed to update apply state", append(apply.MutableLogAttrs(), "error", err)...)
 	} else {
-		// Record the failure in the apply's own log stream. It is the only
-		// surface an operator reads from the CLI or the PR summary, so a failure
-		// that lands only in the server logs reads there as an apply that went
-		// terminal for no stated reason.
-		c.logApplyEvent(ctx, apply.ID, nil, storage.LogLevelError, storage.LogEventError, storage.LogSourceSchemaBot,
-			fmt.Sprintf("Apply failed: %s", errMsg), previousState, state.Apply.Failed)
+		c.logApplyFailure(ctx, apply, previousState, errMsg)
 	}
 	metrics.AdjustActiveApplies(ctx, -1, apply.Database, apply.Deployment, apply.Environment)
 }
@@ -91,14 +108,7 @@ func (c *LocalClient) markApplyRetryableWithTasks(ctx context.Context, apply *st
 	if err := c.storage.Applies().Update(ctx, apply); err != nil {
 		logger.Error("failed to update apply state", append(apply.MutableLogAttrs(), "error", err)...)
 	} else {
-		// Each paused attempt is recorded with the budget it spent, so the apply
-		// log shows recovery burning through its attempts rather than only the
-		// silence between them. The count is the apply's own attempt counter,
-		// which a recovery claim advances — reporting the drive number instead
-		// would run past the budget it is measured against.
-		c.logApplyEvent(ctx, apply.ID, nil, storage.LogLevelWarn, storage.LogEventError, storage.LogSourceSchemaBot,
-			fmt.Sprintf("Apply paused for operator retry (%d of %d recovery attempts used): %s", apply.Attempt, storage.MaxRecoveryAttempts, errMsg),
-			previousState, state.Apply.FailedRetryable)
+		c.logApplyPausedForRetry(ctx, apply, previousState, errMsg)
 	}
 	metrics.AdjustActiveApplies(ctx, -1, apply.Database, apply.Deployment, apply.Environment)
 	if obs := c.getObserver(apply.ID); obs != nil {
diff --git a/pkg/tern/local_apply_sequential.go b/pkg/tern/local_apply_sequential.go
index 1e7be72c4..1dbb8f3fe 100644
--- a/pkg/tern/local_apply_sequential.go
+++ b/pkg/tern/local_apply_sequential.go
@@ -636,6 +636,7 @@ func (c *LocalClient) finalizeSequentialApply(ctx context.Context, apply *storag
 		}
 		return
 	}
+	previousState := apply.State
 	switch {
 	case failedTask != nil && failedTask.State == state.Task.FailedRetryable:
 		apply.State = state.Apply.FailedRetryable
@@ -659,6 +660,16 @@ func (c *LocalClient) finalizeSequentialApply(ctx context.Context, apply *storag
 	apply.UpdatedAt = now
 	if err := c.storage.Applies().Update(ctx, apply); err != nil {
 		logger.Error("failed to update apply state", append(apply.MutableLogAttrs(), "error", err)...)
+	} else {
+		// A sequential apply's failure reaches the operator through the same
+		// apply log stream as every other path, so a failed table does not read
+		// as an apply that went terminal for no stated reason.
+		switch apply.State {
+		case state.Apply.Failed:
+			c.logApplyFailure(ctx, apply, previousState, apply.ErrorMessage)
+		case state.Apply.FailedRetryable:
+			c.logApplyPausedForRetry(ctx, apply, previousState, apply.ErrorMessage)
+		}
 	}
 	if state.IsTerminalApplyState(apply.State) {
 		if err := completePendingRequestsForTerminalApply(ctx, c.storage, apply); err != nil {

From 601a6bc38e2875b39e8964549f0c72bfaa98697f Mon Sep 17 00:00:00 2001
From: Armand Parajon <armand@squareup.com>
Date: Sat, 15 Aug 2026 09:50:54 +0800
Subject: [PATCH 6/6] test(tern): pin the context binding the detached resume's
 log wiring rests on
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A resume that hands its polling to a goroutine outliving the caller rewires the
engine's apply-log callback on that goroutine's own context. The reason is now
covered: a callback still holding the caller's cancelled context records
nothing, so the engine lines for the rest of the schema change would be lost.

The apply log store fixture stops ignoring its context, which is what makes the
difference observable — the real store treats a cancelled context as a failed
write, not a silently dropped one.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 pkg/tern/grpc_client_test.go                |  7 ++++-
 pkg/tern/local_apply_spirit_logging_test.go | 34 +++++++++++++++++++++
 2 files changed, 40 insertions(+), 1 deletion(-)

diff --git a/pkg/tern/grpc_client_test.go b/pkg/tern/grpc_client_test.go
index bdeafacf7..db130e912 100644
--- a/pkg/tern/grpc_client_test.go
+++ b/pkg/tern/grpc_client_test.go
@@ -820,7 +820,12 @@ type mockApplyLogStore struct {
 	recentLimit int
 }
 
-func (m *mockApplyLogStore) Append(_ context.Context, log *storage.ApplyLog) error {
+// Append records the line the way the real store does: a cancelled context is
+// a failed write, not a silently dropped one.
+func (m *mockApplyLogStore) Append(ctx context.Context, log *storage.ApplyLog) error {
+	if err := ctx.Err(); err != nil {
+		return fmt.Errorf("append apply log for apply %d: %w", log.ApplyID, err)
+	}
 	stored := *log
 	m.logs = append(m.logs, &stored)
 	return nil
diff --git a/pkg/tern/local_apply_spirit_logging_test.go b/pkg/tern/local_apply_spirit_logging_test.go
index e3a1921fb..b007d6c55 100644
--- a/pkg/tern/local_apply_spirit_logging_test.go
+++ b/pkg/tern/local_apply_spirit_logging_test.go
@@ -1,6 +1,7 @@
 package tern
 
 import (
+	"context"
 	"log/slog"
 	"testing"
 
@@ -73,3 +74,36 @@ func TestSpiritApplyLogFunc_MultiTableLineAttributedToApply(t *testing.T) {
 	assert.Equal(t, "[drinks, orders] apply complete", logs.logs[0].Message)
 	assert.Nil(t, logs.logs[0].TaskID)
 }
+
+// The engine's log wiring is bound to the context it was built on. A resume
+// that hands its polling to a goroutine outliving the caller therefore rewires
+// on that goroutine's own context: a callback still holding the caller's
+// cancelled context records nothing, and the engine lines for the rest of the
+// schema change — the ones an operator reads when it fails — are exactly the
+// ones that would be lost.
+func TestSpiritApplyLogFunc_BoundContextDecidesWhatSurvivesTheCaller(t *testing.T) {
+	logs := &mockApplyLogStore{}
+	client := &LocalClient{
+		storage: &mockStorage{logs: logs},
+		logger:  slog.Default(),
+	}
+
+	apply := &storage.Apply{ID: 3}
+	tasks := []*storage.Task{{ID: 11, TableName: "orders"}}
+
+	callerCtx, cancelCaller := context.WithCancel(t.Context())
+	boundToCaller := client.spiritApplyLogFunc(callerCtx, apply, tasks)
+	boundToDetachedPoll := client.spiritApplyLogFunc(context.WithoutCancel(callerCtx), apply, tasks)
+
+	cancelCaller()
+
+	boundToCaller(slog.LevelInfo, "orders", "copy rows complete")
+	assert.Empty(t, logs.logs, "a callback holding the caller's cancelled context records nothing")
+
+	boundToDetachedPoll(slog.LevelError, "orders", "fatal error processing GTID rows event")
+	require.Len(t, logs.logs, 1)
+	assert.Equal(t, "[orders] fatal error processing GTID rows event", logs.logs[0].Message)
+	assert.Equal(t, storage.LogLevelError, logs.logs[0].Level)
+	require.NotNil(t, logs.logs[0].TaskID)
+	assert.Equal(t, tasks[0].ID, *logs.logs[0].TaskID)
+}