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 01969181b..bc7bb3530 100644
--- a/pkg/api/operator.go
+++ b/pkg/api/operator.go
@@ -238,6 +238,47 @@ 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) {
+	s.appendApplyLog(ctx, s.logger, &storage.ApplyLog{
+		ApplyID:   apply.ID,
+		Level:     storage.LogLevelError,
+		EventType: storage.LogEventError,
+		Source:    storage.LogSourceSchemaBot,
+		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(),
+	}, "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)...)
 	}
 }
 
@@ -1416,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,
@@ -1432,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
@@ -1548,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,
@@ -1566,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 b9ee2c594..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")
@@ -242,6 +243,40 @@ 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,
+		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)
+}
+
 // 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..c4889fbdc
--- /dev/null
+++ b/pkg/tern/apply_failure_log_test.go
@@ -0,0 +1,175 @@
+package tern
+
+import (
+	"context"
+	"fmt"
+	"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 has left. Without it the apply log shows
+// only the gaps between attempts, and the retry budget drains invisibly. The
+// 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})
+
+	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,
+		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 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)
+
+	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("0 of %d recovery attempts remaining", 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.
+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)
+}
+
+// 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/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_failure.go b/pkg/tern/local_apply_failure.go
index 2dd880b8e..6e40cb313 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"
@@ -9,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
@@ -38,12 +61,15 @@ 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 {
+		c.logApplyFailure(ctx, apply, previousState, errMsg)
 	}
 	metrics.AdjustActiveApplies(ctx, -1, apply.Database, apply.Deployment, apply.Environment)
 }
@@ -74,12 +100,15 @@ 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 {
+		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_grouped.go b/pkg/tern/local_apply_grouped.go
index dd5c74b92..7d6ca1f80 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)
 		return
 	}
 
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 {
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)
+}
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 80f466f1e..229e9ddf8 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,16 @@ 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()
+		defer stopEngineLogging()
 		c.pollForCompletionAtomic(resumeCtx, apply, tasks, creds, resumeState, options, releaseAtCutoverBarrier)
 	}()
 	return nil
@@ -1510,14 +1536,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) {
@@ -1565,8 +1596,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
@@ -1650,8 +1679,6 @@ 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())
 	if startRequested {
 		if failErr := failPendingControlRequests(ctx, c.storage, apply, storage.ControlOperationStart, err.Error()); failErr != nil {
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")
+}