feat(tern): one deployment correlates to exactly one remote apply

pkg/api/handlers_test.go

@@ -953,6 +953,43 @@ func TestHandleDeploymentLogsUsesParentExternalIDForSingleOperation(t *testing.T
 	assert.Equal(t, "apply complete", response.Sources[0].Logs[0].Message)
 }
 
+// A remotely dispatched operation whose remote apply id lives only in the
+// legacy engine resume context carrier still contributes a data-plane log
+// source: the deployment routes to a remote client, so the carrier holds a
+// remote apply id, not engine resume state, and its logs must not be silently
+// missing from the fan-out.
+func TestHandleDeploymentLogsHonorsLegacyResumeContextCarrier(t *testing.T) {
+	apply := &storage.Apply{ID: 9, ApplyIdentifier: "apply-control", Database: "commerce", DatabaseType: storage.DatabaseTypeStrata, Environment: "staging"}
+	operations := []*storage.ApplyOperation{
+		{ApplyID: apply.ID, Deployment: "region-a", OperationKey: "commerce/-80/orders", OperationKind: storage.ApplyOperationKindWork, Target: "cluster-a", ExternalID: "remote-a"},
+		{ApplyID: apply.ID, Deployment: "region-a", OperationKey: "commerce/80-/orders", OperationKind: storage.ApplyOperationKindWork, Target: "cluster-a", EngineResumeContext: "remote-legacy"},
+	}
+	client := &mockTernClient{isRemote: true}
+	client.logsHook = func(req *ternv1.LogsRequest) (*ternv1.LogsResponse, error) {
+		return &ternv1.LogsResponse{ApplyId: req.ApplyId, Logs: []*ternv1.ApplyLog{{Id: 21, Level: "info", Message: "copying", CreatedAt: "2026-07-18T18:33:10Z"}}}, nil
+	}
+	service := New(&mockStorageWithApplyStores{
+		applies:    &staticApplyStore{apply: apply},
+		operations: &staticApplyOperationStore{operations: operations},
+	}, testServerConfig(), map[string]tern.Client{"region-a/staging": client}, slog.New(slog.NewTextHandler(io.Discard, nil)))
+
+	req := httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/api/logs?apply_id=apply-control&deployment=region-a", nil)
+	w := httptest.NewRecorder()
+	service.handleLogsWithoutDatabase(w, req)
+
+	require.Equal(t, http.StatusOK, w.Code)
+	require.Len(t, client.logsReqs, 2)
+	assert.Equal(t, "remote-a", client.logsReqs[0].ApplyId)
+	assert.Equal(t, "remote-legacy", client.logsReqs[1].ApplyId)
+	var response apitypes.DeploymentLogsResponse
+	require.NoError(t, json.Unmarshal(w.Body.Bytes(), &response))
+	require.Len(t, response.Sources, 2)
+	assert.Equal(t, "remote-a", response.Sources[0].ExternalID)
+	assert.Equal(t, "remote-legacy", response.Sources[1].ExternalID)
+	require.Len(t, response.Sources[1].Operations, 1)
+	assert.Equal(t, "commerce/80-/orders", response.Sources[1].Operations[0].OperationKey)
+}
+
 func (m *mockTernClient) Cutover(ctx context.Context, req *ternv1.CutoverRequest) (*ternv1.CutoverResponse, error) {
 	m.cutoverReq = req
 	if m.cutoverResp != nil {

pkg/api/log_handlers.go

@@ -159,20 +159,46 @@ func (s *Service) handleDeploymentLogs(w http.ResponseWriter, r *http.Request, a
 		s.writeError(w, http.StatusInternalServerError, "failed to list apply operations")
 		return
 	}
-	fetches := make(map[string]*deploymentLogFetch)
 	matched := false
+	for _, op := range ops {
+		if op.Deployment == deployment {
+			matched = true
+			break
+		}
+	}
+	if !matched {
+		s.writeError(w, http.StatusBadRequest, fmt.Sprintf("deployment %q has no operations for apply %q", deployment, apply.ApplyIdentifier))
+		return
+	}
+	client, err := s.TernClient(deployment, apply.Environment)
+	if err != nil {
+		s.logger.Error("failed to resolve deployment for data-plane logs",
+			append(apply.LogAttrs(),
+				"operation", "read_deployment_logs", "operation_deployment", deployment, "error", err)...)
+		s.writeError(w, http.StatusBadRequest, fmt.Sprintf("cannot resolve deployment %q; check server logs", deployment))
+		return
+	}
+	if !client.IsRemote() {
+		s.writeError(w, http.StatusBadRequest, fmt.Sprintf("deployment %q is local-only; omit --deployment to read control-plane logs", deployment))
+		return
+	}
+	fetches := make(map[string]*deploymentLogFetch)
 	for _, op := range ops {
 		if op.Deployment != deployment {
 			continue
 		}
-		matched = true
-		externalID := op.ExternalID
+		// The deployment is proven remote above, so the operation's recorded
+		// remote apply id — including one living only in the legacy engine
+		// resume context carrier — is a data-plane apply id, not engine resume
+		// state.
+		externalID := op.RemoteApplyID()
 		if externalID == "" && len(ops) == 1 {
 			externalID = apply.ExternalID
 		}
 		if externalID == "" {
-			// An operation without a remote apply id ran on the control plane;
-			// its logs live in control-plane storage, not behind this fan-out.
+			// An operation without a remote apply id has not been dispatched to
+			// the data plane; its logs live in control-plane storage, not behind
+			// this fan-out.
 			s.logger.Debug("skipping operation without a remote apply id for data-plane logs",
 				append(apply.LogAttrs(),
 					"operation", "read_deployment_logs", "operation_deployment", deployment, "operation_key", op.OperationKey, "target", op.Target)...)
@@ -186,26 +212,10 @@ func (s *Service) handleDeploymentLogs(w http.ResponseWriter, r *http.Request, a
 		}
 		fetch.operations = append(fetch.operations, &apitypes.LogOperationProvenance{OperationKey: op.OperationKey, Target: op.Target, OperationKind: op.OperationKind})
 	}
-	if !matched {
-		s.writeError(w, http.StatusBadRequest, fmt.Sprintf("deployment %q has no operations for apply %q", deployment, apply.ApplyIdentifier))
-		return
-	}
 	if len(fetches) == 0 {
 		s.writeError(w, http.StatusBadRequest, fmt.Sprintf("deployment %q has no remote operation logs; omit --deployment to read control-plane logs", deployment))
 		return
 	}
-	client, err := s.TernClient(deployment, apply.Environment)
-	if err != nil {
-		s.logger.Error("failed to resolve deployment for data-plane logs",
-			append(apply.LogAttrs(),
-				"operation", "read_deployment_logs", "operation_deployment", deployment, "error", err)...)
-		s.writeError(w, http.StatusBadRequest, fmt.Sprintf("cannot resolve deployment %q; check server logs", deployment))
-		return
-	}
-	if !client.IsRemote() {
-		s.writeError(w, http.StatusBadRequest, fmt.Sprintf("deployment %q is local-only; omit --deployment to read control-plane logs", deployment))
-		return
-	}
 	result := &apitypes.DeploymentLogsResponse{ApplyID: apply.ApplyIdentifier, Deployment: deployment, Sources: []*apitypes.DeploymentLogSource{}, Errors: []*apitypes.DeploymentLogError{}}
 	keys := make([]string, 0, len(fetches))
 	for key := range fetches {

pkg/metrics/README.md

@@ -45,6 +45,7 @@ available, such as `repository`, `github_app`, and `installation_id`.
 | `schemabot.remote_apply_dedup_total` | Counter | database, environment, outcome | Idempotency-keyed dispatches replayed against their existing operation on the keyed apply — see [Remote Apply Attaches](#remote-apply-attaches) |
 | `schemabot.remote_apply_attach_total` | Counter | database, environment, outcome | Sibling dispatches resolved into an existing deployment-keyed apply — see [Remote Apply Attaches](#remote-apply-attaches) |
 | `schemabot.remote_apply_key_echo_mismatch_total` | Counter | database, environment | Remote dispatches refused fail-closed because the data plane's accepted response echoed a different operation key than the dispatch derives — see [Remote Apply Attaches](#remote-apply-attaches) |
+| `schemabot.remote_apply_deployment_id_conflict_total` | Counter | database, environment, deployment | Remote dispatch results refused fail-closed because the deployment already correlates to a different remote apply id — see [Remote Apply Attaches](#remote-apply-attaches) |
 | `schemabot.lock_operations_total` | Counter | operation, database, environment, status | Lock acquire/release operations |
 | `schemabot.direct_write_authorization.total` | Counter | operation, database, environment, status, reason | Per-database direct-write (CLI/API) authorization decisions at the handler layer |
 | `schemabot.operator.resumed_total` | Counter | database, environment, previous_state | Applies resumed by the operator |
@@ -346,6 +347,17 @@ data plane so both planes derive the same key, then retry the blocked applies;
 the paired error log carries the apply, the dispatched operation key, and the
 echoed key.
 
+`schemabot.remote_apply_deployment_id_conflict_total` counts dispatch results
+the control plane refused because storing them would correlate one deployment
+to two remote applies. All operations of a deployment attach into the
+deployment's single data-plane apply and record the same remote apply id, so a
+second id means the planes diverged — an in-flight apply spanning a
+dispatch-key rollout, or a data plane that lost its keyed apply and minted a
+fresh one. The operator action is to inspect the named database's apply
+operations (the paired error log carries the recorded and refused ids), decide
+which remote apply is authoritative, and re-dispatch under a fresh generation
+once the planes agree.
+
 `schemabot.lock_operations_total` tracks database-level lock acquisition and
 release attempts.
 

pkg/metrics/metrics.go

@@ -972,6 +972,24 @@ func RecordRemoteApplyKeyEchoMismatch(ctx context.Context, database, environment
 	)
 }
 
+// RecordRemoteApplyDeploymentIDConflict increments the counter for remote
+// dispatch results refused because the deployment already correlates to a
+// different remote apply id. One deployment maps to exactly one data-plane
+// apply, so a second id means the planes have diverged — an in-flight apply
+// spanning a dispatch-key rollout, or a data plane that lost its keyed apply
+// and minted a fresh one. The refusal fails the dispatch closed; the operator
+// action is to inspect the named database's apply operations, decide which
+// remote apply is authoritative, and re-dispatch under a fresh generation once
+// the planes agree.
+func RecordRemoteApplyDeploymentIDConflict(ctx context.Context, database, environment, deployment string) {
+	addCounter(ctx, "schemabot.remote_apply_deployment_id_conflict_total",
+		"Total remote apply dispatches refused because the deployment already correlates to a different remote apply id", "{dispatch}",
+		attribute.String("database", database),
+		EnvironmentAttribute(environment),
+		attribute.String("deployment", deployment),
+	)
+}
+
 // operatorMetricNames returns the canonical operator metric name alongside its
 // deprecated schemabot.scheduler.* alias. Both are emitted for one release so
 // dashboards and alerts can migrate before the legacy series is removed.

pkg/storage/deployment_remote_apply.go

@@ -0,0 +1,54 @@
+package storage
+
+import "fmt"
+
+// RemoteApplyID returns the remote data-plane apply identifier recorded on
+// this operation row, or "" when none has been recorded yet. external_id is
+// the canonical column; engine_resume_context is the legacy carrier remote
+// drives persisted the id into before the external-id columns existed, so
+// readers fall back to it. Meaningful only for operations dispatched to a
+// remote (gRPC) data plane — on locally driven operations
+// engine_resume_context holds engine-owned resume state, not an apply id, so
+// callers must gate on the dispatch shape before treating the result as a
+// remote apply id.
+func (op *ApplyOperation) RemoteApplyID() string {
+	if op == nil {
+		return ""
+	}
+	if op.ExternalID != "" {
+		return op.ExternalID
+	}
+	return op.EngineResumeContext
+}
+
+// DeploymentRemoteApplyID resolves the single remote data-plane apply id
+// shared by one deployment's operations of an apply. A remote deployment has
+// exactly one data-plane apply: every operation dispatched for it attaches
+// into that apply, so every operation row of the deployment records the same
+// remote apply id. It returns "" when no operation of the deployment has
+// recorded one yet, and an error when the deployment's operations disagree —
+// two remote apply ids for one deployment means the planes have diverged
+// (an in-flight apply spanning a dispatch-key rollout, or a data plane that
+// lost its keyed apply) and callers must fail closed rather than pick one.
+// Operations of other deployments are ignored: sibling deployments of the
+// same apply legitimately carry their own distinct remote apply ids.
+func DeploymentRemoteApplyID(ops []*ApplyOperation, deployment string) (string, error) {
+	shared := ""
+	for _, op := range ops {
+		if op == nil || op.Deployment != deployment {
+			continue
+		}
+		id := op.RemoteApplyID()
+		if id == "" {
+			continue
+		}
+		if shared == "" {
+			shared = id
+			continue
+		}
+		if id != shared {
+			return "", fmt.Errorf("deployment %q operations record more than one remote apply id (%q on apply_operation %d disagrees with %q)", deployment, id, op.ID, shared)
+		}
+	}
+	return shared, nil
+}

pkg/storage/deployment_remote_apply_test.go

@@ -0,0 +1,89 @@
+package storage
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestApplyOperationRemoteApplyID(t *testing.T) {
+	t.Run("nil operation", func(t *testing.T) {
+		var op *ApplyOperation
+		assert.Empty(t, op.RemoteApplyID())
+	})
+	t.Run("external id is canonical", func(t *testing.T) {
+		op := &ApplyOperation{ExternalID: "apply-remote-1", EngineResumeContext: "legacy-ctx"}
+		assert.Equal(t, "apply-remote-1", op.RemoteApplyID())
+	})
+	t.Run("legacy resume context carrier", func(t *testing.T) {
+		op := &ApplyOperation{EngineResumeContext: "apply-legacy-1"}
+		assert.Equal(t, "apply-legacy-1", op.RemoteApplyID())
+	})
+	t.Run("nothing recorded", func(t *testing.T) {
+		assert.Empty(t, (&ApplyOperation{}).RemoteApplyID())
+	})
+}
+
+func TestDeploymentRemoteApplyID(t *testing.T) {
+	t.Run("no operations", func(t *testing.T) {
+		id, err := DeploymentRemoteApplyID(nil, "west")
+		require.NoError(t, err)
+		assert.Empty(t, id)
+	})
+
+	t.Run("nothing recorded yet", func(t *testing.T) {
+		ops := []*ApplyOperation{
+			{ID: 1, Deployment: "west", OperationKey: "ns/-80/users"},
+			{ID: 2, Deployment: "west", OperationKey: "ns/80-/users"},
+		}
+		id, err := DeploymentRemoteApplyID(ops, "west")
+		require.NoError(t, err)
+		assert.Empty(t, id)
+	})
+
+	t.Run("all siblings agree", func(t *testing.T) {
+		ops := []*ApplyOperation{
+			{ID: 1, Deployment: "west", ExternalID: "apply-remote-1"},
+			{ID: 2, Deployment: "west", ExternalID: "apply-remote-1"},
+			{ID: 3, Deployment: "west"},
+		}
+		id, err := DeploymentRemoteApplyID(ops, "west")
+		require.NoError(t, err)
+		assert.Equal(t, "apply-remote-1", id)
+	})
+
+	t.Run("legacy carrier counts as the recorded id", func(t *testing.T) {
+		ops := []*ApplyOperation{
+			{ID: 1, Deployment: "west", EngineResumeContext: "apply-remote-1"},
+			{ID: 2, Deployment: "west", ExternalID: "apply-remote-1"},
+		}
+		id, err := DeploymentRemoteApplyID(ops, "west")
+		require.NoError(t, err)
+		assert.Equal(t, "apply-remote-1", id)
+	})
+
+	t.Run("sibling deployments keep their own remote applies", func(t *testing.T) {
+		ops := []*ApplyOperation{
+			{ID: 1, Deployment: "west", ExternalID: "apply-remote-west"},
+			{ID: 2, Deployment: "east", ExternalID: "apply-remote-east"},
+			{ID: 3, Deployment: "south", ExternalID: "apply-remote-south"},
+		}
+		id, err := DeploymentRemoteApplyID(ops, "east")
+		require.NoError(t, err)
+		assert.Equal(t, "apply-remote-east", id)
+	})
+
+	t.Run("disagreeing siblings fail closed", func(t *testing.T) {
+		ops := []*ApplyOperation{
+			{ID: 1, Deployment: "west", ExternalID: "apply-remote-1"},
+			{ID: 2, Deployment: "west", ExternalID: "apply-remote-2"},
+		}
+		id, err := DeploymentRemoteApplyID(ops, "west")
+		require.Error(t, err)
+		assert.Empty(t, id)
+		assert.Contains(t, err.Error(), "apply-remote-1")
+		assert.Contains(t, err.Error(), "apply-remote-2")
+		assert.Contains(t, err.Error(), `deployment "west"`)
+	})
+}

pkg/storage/errors.go

@@ -58,6 +58,13 @@ var (
 	// row does not exist for the given lookup key.
 	ErrApplyOperationNotFound = errors.New("apply operation not found")
 
+	// ErrRemoteApplyDeploymentIDConflict is returned when storing a remote
+	// apply id would correlate one deployment to more than one remote
+	// data-plane apply — either the deployment's operations already disagree
+	// with each other, or the id being stored disagrees with the one they
+	// share. Callers must fail closed rather than pick one.
+	ErrRemoteApplyDeploymentIDConflict = errors.New("deployment already correlates to a different remote apply")
+
 	// ErrApplyOperationExists is returned when an apply_operations row for
 	// (apply_id, deployment, operation_key) is being inserted but already exists.
 	ErrApplyOperationExists = errors.New("apply operation already exists")