diff --git a/pkg/actions/actions.go b/pkg/actions/actions.go
index c7ef749df..9b35bb55a 100644
--- a/pkg/actions/actions.go
+++ b/pkg/actions/actions.go
@@ -636,6 +636,7 @@ func (a *ActionManager) invokeGlobalAction(
 		}()
 		oa.SetStatus(ctx, v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING)
 		bgCtx := trace.ContextWithSpanContext(context.Background(), trace.SpanContextFromContext(ctx))
+		bgCtx = ctxzap.ToContext(bgCtx, ctxzap.Extract(ctx))
 		handlerCtx, cancel := context.WithTimeoutCause(bgCtx, 1*time.Hour, errors.New("action handler timed out"))
 		defer cancel()
 		rv, annos, oaErr := handler(handlerCtx, args)
diff --git a/pkg/connectorbuilder/actions.go b/pkg/connectorbuilder/actions.go
index 314bdb782..8f0038c6b 100644
--- a/pkg/connectorbuilder/actions.go
+++ b/pkg/connectorbuilder/actions.go
@@ -5,14 +5,39 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/grpc-ecosystem/go-grpc-middleware/logging/zap/ctxzap"
+	"go.uber.org/zap"
+	"google.golang.org/protobuf/types/known/structpb"
+
 	v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2"
 	"github.com/conductorone/baton-sdk/pkg/actions"
 	"github.com/conductorone/baton-sdk/pkg/annotations"
 	"github.com/conductorone/baton-sdk/pkg/types/tasks"
 	"github.com/conductorone/baton-sdk/pkg/uotel"
-	"google.golang.org/protobuf/types/known/structpb"
 )
 
+const (
+	// maxConsecutiveStatusErrors bounds how many status-check failures or
+	// indeterminate statuses in a row the legacy action poll loop tolerates
+	// before failing the action.
+	maxConsecutiveStatusErrors = 3
+)
+
+// legacyPollIntervals paces the legacy status poll: it starts fast and backs
+// off to a cap so a slow action doesn't drain a remote manager's rate-limit
+// budget. Captured at registration, so tests can drive the loop without
+// real-time waits and the detached poll goroutine never reads shared
+// mutable state.
+type legacyPollIntervals struct {
+	initial time.Duration
+	max     time.Duration
+}
+
+var defaultLegacyPollIntervals = legacyPollIntervals{
+	initial: time.Second,
+	max:     30 * time.Second,
+}
+
 // ActionManager defines the interface for managing actions in the connector builder.
 // This is the internal interface used by the builder for dispatch.
 // The *actions.ActionManager type implements this interface.
@@ -210,14 +235,139 @@ func (b *builder) GetActionStatus(ctx context.Context, request *v2.GetActionStat
 }
 
 // registerLegacyAction wraps a legacy CustomActionManager action as an ActionHandler and registers it.
-func registerLegacyAction(ctx context.Context, registry actions.ActionRegistry, schema *v2.BatonActionSchema, legacyManager CustomActionManager) error {
+func registerLegacyAction(
+	ctx context.Context,
+	registry actions.ActionRegistry,
+	schema *v2.BatonActionSchema,
+	legacyManager CustomActionManager,
+	intervals legacyPollIntervals,
+) error {
+	// A zero or negative interval would busy-loop the status poll. An
+	// inverted pair needs no guard: the cap applies from the second tick.
+	if intervals.initial <= 0 || intervals.max <= 0 {
+		intervals = defaultLegacyPollIntervals
+	}
 	handler := func(ctx context.Context, args *structpb.Struct) (*structpb.Struct, annotations.Annotations, error) {
-		_, _, resp, annos, err := legacyManager.InvokeAction(ctx, schema.GetName(), "", args)
-		return resp, annos, err
+		// The inner call keeps the detached handler context; its one-hour
+		// deadline is the execution backstop for however long the legacy
+		// manager runs.
+		id, actionStatus, resp, annos, err := legacyManager.InvokeAction(ctx, schema.GetName(), "", args)
+		if err != nil {
+			return resp, annos, err
+		}
+
+		// Legacy managers were never required to populate id or status — the
+		// wrapper used to discard both — so the never-populated shape
+		// resolves the outer action with the response, as it always did.
+		if actionStatus == v2.BatonActionStatus_BATON_ACTION_STATUS_UNSPECIFIED {
+			return resp, annos, nil
+		}
+
+		// A settled status at the invoke seam resolves like a terminal
+		// poll. The SDK's own manager reports handler failures in-band as
+		// FAILED with a nil error, so this must not resolve as success.
+		if isSettledActionStatus(actionStatus) {
+			return resp, annos, legacyStatusErr(schema.GetName(), actionStatus, resp)
+		}
+
+		// An unresolved claim without an id cannot be polled; resolve with
+		// the response, matching the old fire-and-forget behavior.
+		if id == "" {
+			return resp, annos, nil
+		}
+
+		// In-flight and indeterminate statuses alike are polled: an
+		// indeterminate answer gets the same tolerance here as one arriving
+		// from a later poll.
+
+		// Poll to a terminal status so the outer result carries the action's
+		// real outcome. A few consecutive status-check failures are tolerated:
+		// one flaky remote lookup must not convert a succeeding action into a
+		// failure. The interval backs off to a cap so a slow action doesn't
+		// drain a remote manager's rate-limit budget.
+		l := ctxzap.Extract(ctx)
+		statusErrs := 0
+		interval := intervals.initial
+		timer := time.NewTimer(interval)
+		defer timer.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return resp, annos, fmt.Errorf("legacy action %q did not reach a terminal status: %w", schema.GetName(), context.Cause(ctx))
+			case <-timer.C:
+			}
+			interval = min(interval*2, intervals.max)
+			timer.Reset(interval)
+
+			st, _, pollResp, pollAnnos, err := legacyManager.GetActionStatus(ctx, id)
+			if err != nil {
+				statusErrs++
+				if statusErrs >= maxConsecutiveStatusErrors {
+					return resp, annos, fmt.Errorf("legacy action %q status lookup failed: %w", schema.GetName(), err)
+				}
+				l.Warn("legacy action status check failed, retrying",
+					zap.String("action", schema.GetName()),
+					zap.Int("consecutive_anomalies", statusErrs),
+					zap.Error(err))
+				continue
+			}
+			// Keep the last meaningful response for the error exits above;
+			// an indeterminate poll's payload must not replace it.
+			if pollResp != nil && (isInFlightActionStatus(st) || isSettledActionStatus(st)) {
+				resp, annos = pollResp, pollAnnos
+			}
+
+			switch {
+			case isInFlightActionStatus(st):
+				statusErrs = 0
+			case isSettledActionStatus(st):
+				// The settling poll's own payload is the failure account;
+				// resp may hold an older in-flight snapshot.
+				return resp, annos, legacyStatusErr(schema.GetName(), st, pollResp)
+			default:
+				// An indeterminate status gets the same tolerance as a
+				// lookup error: transient anomalies recover, persistent
+				// ones fail closed.
+				statusErrs++
+				if statusErrs >= maxConsecutiveStatusErrors {
+					return resp, annos, fmt.Errorf("legacy action %q returned unexpected status %s", schema.GetName(), st.String())
+				}
+				l.Warn("legacy action returned indeterminate status, retrying",
+					zap.String("action", schema.GetName()),
+					zap.String("status", st.String()),
+					zap.Int("consecutive_anomalies", statusErrs))
+			}
+		}
 	}
 	return registry.Register(ctx, schema, handler)
 }
 
+// legacyStatusErr maps a settled legacy status — COMPLETE or FAILED, the
+// only values both call sites pass — to the outer handler error, carrying
+// the error message the settling response reports, since the outer error
+// replaces the response's error field.
+func legacyStatusErr(name string, st v2.BatonActionStatus, resp *structpb.Struct) error {
+	if st == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE {
+		return nil
+	}
+	if inner := resp.GetFields()["error"].GetStringValue(); inner != "" {
+		return fmt.Errorf("legacy action %q failed: %s", name, inner)
+	}
+	return fmt.Errorf("legacy action %q failed", name)
+}
+
+func isInFlightActionStatus(s v2.BatonActionStatus) bool {
+	return s == v2.BatonActionStatus_BATON_ACTION_STATUS_PENDING || s == v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING
+}
+
+// isSettledActionStatus is the single gate deciding which statuses resolve
+// immediately, at the invoke seam and from polls alike. A new terminal enum
+// value must be added here, or it takes the indeterminate path: polled to
+// the tolerance threshold, then failed closed.
+func isSettledActionStatus(s v2.BatonActionStatus) bool {
+	return s == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE || s == v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED
+}
+
 // addActionManager handles deprecated CustomActionManager and RegisterActionManagerLimited interfaces
 // by extracting their actions and registering them into the unified ActionManager.
 func (b *builder) addActionManager(ctx context.Context, in interface{}, registry actions.ActionRegistry) error {
@@ -228,7 +378,7 @@ func (b *builder) addActionManager(ctx context.Context, in interface{}, registry
 			return fmt.Errorf("error listing schemas from custom action manager: %w", err)
 		}
 		for _, schema := range schemas {
-			if err := registerLegacyAction(ctx, registry, schema, customManager); err != nil {
+			if err := registerLegacyAction(ctx, registry, schema, customManager, defaultLegacyPollIntervals); err != nil {
 				return fmt.Errorf("error registering legacy action %s: %w", schema.GetName(), err)
 			}
 		}
@@ -249,7 +399,7 @@ func (b *builder) addActionManager(ctx context.Context, in interface{}, registry
 			return fmt.Errorf("error listing schemas from custom action manager: %w", err)
 		}
 		for _, schema := range schemas {
-			if err := registerLegacyAction(ctx, registry, schema, customManager); err != nil {
+			if err := registerLegacyAction(ctx, registry, schema, customManager, defaultLegacyPollIntervals); err != nil {
 				return fmt.Errorf("error registering legacy action %s: %w", schema.GetName(), err)
 			}
 		}
diff --git a/pkg/connectorbuilder/actions_legacy_test.go b/pkg/connectorbuilder/actions_legacy_test.go
new file mode 100644
index 000000000..2e6170b4d
--- /dev/null
+++ b/pkg/connectorbuilder/actions_legacy_test.go
@@ -0,0 +1,550 @@
+package connectorbuilder
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"slices"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2"
+	"github.com/conductorone/baton-sdk/pkg/actions"
+	"github.com/conductorone/baton-sdk/pkg/annotations"
+	"github.com/grpc-ecosystem/go-grpc-middleware/logging/zap/ctxzap"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zaptest/observer"
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+type fakeThirdPartyActionManager struct {
+	schema         *v2.BatonActionSchema
+	invokeDeadline time.Time
+	hadDeadline    bool
+	invoked        chan struct{}
+}
+
+func (f *fakeThirdPartyActionManager) ListActionSchemas(_ context.Context, _ string) ([]*v2.BatonActionSchema, annotations.Annotations, error) {
+	return []*v2.BatonActionSchema{f.schema}, nil, nil
+}
+
+func (f *fakeThirdPartyActionManager) GetActionSchema(_ context.Context, _ string) (*v2.BatonActionSchema, annotations.Annotations, error) {
+	return f.schema, nil, nil
+}
+
+func (f *fakeThirdPartyActionManager) InvokeAction(ctx context.Context, _ string, _ string, _ *structpb.Struct) (string, v2.BatonActionStatus, *structpb.Struct, annotations.Annotations, error) {
+	f.invokeDeadline, f.hadDeadline = ctx.Deadline()
+	close(f.invoked)
+	return "legacy-1", v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE, nil, nil, nil
+}
+
+func (f *fakeThirdPartyActionManager) GetActionStatus(_ context.Context, _ string) (v2.BatonActionStatus, string, *structpb.Struct, annotations.Annotations, error) {
+	return v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE, "legacy", nil, nil, nil
+}
+
+// A third-party CustomActionManager must receive the detached handler context
+// (one-hour deadline), not the 2s inline-wait pin reserved for the SDK's own
+// deadline-aware ActionManager: it treats the deadline as an execution cap.
+func TestRegisterLegacyActionThirdPartyManagerKeepsHandlerContext(t *testing.T) {
+	ctx := t.Context()
+
+	legacy := &fakeThirdPartyActionManager{
+		schema:  v2.BatonActionSchema_builder{Name: "legacy_action"}.Build(),
+		invoked: make(chan struct{}),
+	}
+
+	m := actions.NewActionManager(ctx)
+	require.NoError(t, registerLegacyAction(ctx, m, legacy.schema, legacy, shortPollIntervals))
+
+	_, _, _, _, err := m.InvokeAction(ctx, "legacy_action", "", &structpb.Struct{})
+	require.NoError(t, err)
+
+	select {
+	case <-legacy.invoked:
+	case <-time.After(5 * time.Second):
+		t.Fatal("legacy manager was never invoked")
+	}
+
+	require.True(t, legacy.hadDeadline)
+	require.Greater(t, time.Until(legacy.invokeDeadline), 30*time.Minute)
+}
+
+// A deadline-aware inner ActionManager blocks until the action truly
+// finishes: the outer action must stay RUNNING at its own inline wait and
+// resolve later with the real response, never as an empty completion.
+func TestRegisterLegacyActionTracksInnerManagerToCompletion(t *testing.T) {
+	ctx := t.Context()
+
+	schema := v2.BatonActionSchema_builder{Name: "inner_action"}.Build()
+	rv, err := structpb.NewStruct(map[string]any{"done": true})
+	require.NoError(t, err)
+
+	inner := actions.NewActionManager(ctx)
+	require.NoError(t, inner.Register(ctx, schema, func(_ context.Context, _ *structpb.Struct) (*structpb.Struct, annotations.Annotations, error) {
+		time.Sleep(1500 * time.Millisecond)
+		return rv, nil, nil
+	}))
+
+	outer := actions.NewActionManager(ctx)
+	require.NoError(t, registerLegacyAction(ctx, outer, schema, inner, shortPollIntervals))
+
+	outerID, outerStatus, outerRv, _, err := outer.InvokeAction(ctx, "inner_action", "", &structpb.Struct{})
+	require.NoError(t, err)
+	require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING, outerStatus)
+	require.Nil(t, outerRv)
+
+	require.Eventually(t, func() bool {
+		st, _, gotRv, _, err := outer.GetActionStatus(ctx, outerID)
+		return err == nil && st == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE && gotRv != nil
+	}, 5*time.Second, 100*time.Millisecond)
+}
+
+type fakeAsyncThirdPartyActionManager struct {
+	schema      *v2.BatonActionSchema
+	rv          *structpb.Struct
+	finalStatus v2.BatonActionStatus
+	statusCalls atomic.Int32
+	gotID       atomic.Value
+}
+
+func (f *fakeAsyncThirdPartyActionManager) ListActionSchemas(_ context.Context, _ string) ([]*v2.BatonActionSchema, annotations.Annotations, error) {
+	return []*v2.BatonActionSchema{f.schema}, nil, nil
+}
+
+func (f *fakeAsyncThirdPartyActionManager) GetActionSchema(_ context.Context, _ string) (*v2.BatonActionSchema, annotations.Annotations, error) {
+	return f.schema, nil, nil
+}
+
+func (f *fakeAsyncThirdPartyActionManager) InvokeAction(_ context.Context, _ string, _ string, _ *structpb.Struct) (string, v2.BatonActionStatus, *structpb.Struct, annotations.Annotations, error) {
+	return "legacy-async-1", v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING, nil, nil, nil
+}
+
+func (f *fakeAsyncThirdPartyActionManager) GetActionStatus(_ context.Context, id string) (v2.BatonActionStatus, string, *structpb.Struct, annotations.Annotations, error) {
+	f.gotID.Store(id)
+	switch f.statusCalls.Add(1) {
+	case 1:
+		// One transient lookup failure must not fail the action.
+		return v2.BatonActionStatus_BATON_ACTION_STATUS_UNKNOWN, "", nil, nil, context.DeadlineExceeded
+	case 2:
+		return v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING, "async", nil, nil, nil
+	default:
+		return f.finalStatus, "async", f.rv, nil, nil
+	}
+}
+
+// A third-party manager returning a non-terminal status gets polled to a
+// terminal one with the action id it returned, riding through a transient
+// status-check failure.
+func TestRegisterLegacyActionPollsAsyncThirdPartyManager(t *testing.T) {
+	ctx := t.Context()
+
+	rv, err := structpb.NewStruct(map[string]any{"done": true})
+	require.NoError(t, err)
+	legacy := &fakeAsyncThirdPartyActionManager{
+		schema:      v2.BatonActionSchema_builder{Name: "async_action"}.Build(),
+		rv:          rv,
+		finalStatus: v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE,
+	}
+
+	outer := actions.NewActionManager(ctx)
+	require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy, shortPollIntervals))
+
+	outerID, _, _, _, err := outer.InvokeAction(ctx, "async_action", "", &structpb.Struct{})
+	require.NoError(t, err)
+
+	require.Eventually(t, func() bool {
+		st, _, gotRv, _, err := outer.GetActionStatus(ctx, outerID)
+		return err == nil && st == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE && gotRv != nil
+	}, 5*time.Second, 10*time.Millisecond)
+
+	require.Equal(t, "legacy-async-1", legacy.gotID.Load())
+	require.GreaterOrEqual(t, legacy.statusCalls.Load(), int32(3))
+}
+
+// A legacy action that polls to FAILED must mark the outer action FAILED,
+// never resolve it as a success.
+func TestRegisterLegacyActionPolledFailureFailsOuterAction(t *testing.T) {
+	ctx := t.Context()
+
+	legacy := &fakeAsyncThirdPartyActionManager{
+		schema:      v2.BatonActionSchema_builder{Name: "failing_action"}.Build(),
+		finalStatus: v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED,
+	}
+
+	outer := actions.NewActionManager(ctx)
+	require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy, shortPollIntervals))
+
+	outerID, _, _, _, err := outer.InvokeAction(ctx, "failing_action", "", &structpb.Struct{})
+	require.NoError(t, err)
+
+	require.Eventually(t, func() bool {
+		st, _, _, _, err := outer.GetActionStatus(ctx, outerID)
+		return err == nil && st == v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED
+	}, 5*time.Second, 10*time.Millisecond)
+}
+
+type fakeSyncNoStatusActionManager struct {
+	schema      *v2.BatonActionSchema
+	rv          *structpb.Struct
+	statusCalls atomic.Int32
+}
+
+func (f *fakeSyncNoStatusActionManager) ListActionSchemas(_ context.Context, _ string) ([]*v2.BatonActionSchema, annotations.Annotations, error) {
+	return []*v2.BatonActionSchema{f.schema}, nil, nil
+}
+
+func (f *fakeSyncNoStatusActionManager) GetActionSchema(_ context.Context, _ string) (*v2.BatonActionSchema, annotations.Annotations, error) {
+	return f.schema, nil, nil
+}
+
+func (f *fakeSyncNoStatusActionManager) InvokeAction(_ context.Context, _ string, _ string, _ *structpb.Struct) (string, v2.BatonActionStatus, *structpb.Struct, annotations.Annotations, error) {
+	return "", v2.BatonActionStatus_BATON_ACTION_STATUS_UNSPECIFIED, f.rv, nil, nil
+}
+
+func (f *fakeSyncNoStatusActionManager) GetActionStatus(_ context.Context, _ string) (v2.BatonActionStatus, string, *structpb.Struct, annotations.Annotations, error) {
+	f.statusCalls.Add(1)
+	return v2.BatonActionStatus_BATON_ACTION_STATUS_UNSPECIFIED, "", nil, nil, nil
+}
+
+// Legacy synchronous managers were never required to populate id or status;
+// a response with the zero status and no id must resolve the outer action
+// immediately, never enter the polling loop.
+func TestRegisterLegacyActionSyncManagerWithoutStatusResolves(t *testing.T) {
+	ctx := t.Context()
+
+	rv, err := structpb.NewStruct(map[string]any{"done": true})
+	require.NoError(t, err)
+	legacy := &fakeSyncNoStatusActionManager{
+		schema: v2.BatonActionSchema_builder{Name: "sync_action"}.Build(),
+		rv:     rv,
+	}
+
+	outer := actions.NewActionManager(ctx)
+	require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy, shortPollIntervals))
+
+	_, outerStatus, outerRv, _, err := outer.InvokeAction(ctx, "sync_action", "", &structpb.Struct{})
+	require.NoError(t, err)
+	require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE, outerStatus)
+	require.NotNil(t, outerRv)
+	require.Equal(t, int32(0), legacy.statusCalls.Load())
+}
+
+// shortPollIntervals drives the poll loop in milliseconds so outcome tables
+// don't add real wall time to the suite.
+var shortPollIntervals = legacyPollIntervals{initial: time.Millisecond, max: 4 * time.Millisecond}
+
+type scriptedPollResult struct {
+	status v2.BatonActionStatus
+	err    error
+	resp   *structpb.Struct
+}
+
+type scriptedLegacyActionManager struct {
+	schema       *v2.BatonActionSchema
+	invokeID     string
+	invokeStatus v2.BatonActionStatus
+	invokeRv     *structpb.Struct
+	polls        []scriptedPollResult
+	pollCalls    atomic.Int32
+}
+
+func (f *scriptedLegacyActionManager) ListActionSchemas(_ context.Context, _ string) ([]*v2.BatonActionSchema, annotations.Annotations, error) {
+	return []*v2.BatonActionSchema{f.schema}, nil, nil
+}
+
+func (f *scriptedLegacyActionManager) GetActionSchema(_ context.Context, _ string) (*v2.BatonActionSchema, annotations.Annotations, error) {
+	return f.schema, nil, nil
+}
+
+func (f *scriptedLegacyActionManager) InvokeAction(_ context.Context, _ string, _ string, _ *structpb.Struct) (string, v2.BatonActionStatus, *structpb.Struct, annotations.Annotations, error) {
+	return f.invokeID, f.invokeStatus, f.invokeRv, nil, nil
+}
+
+func (f *scriptedLegacyActionManager) GetActionStatus(_ context.Context, _ string) (v2.BatonActionStatus, string, *structpb.Struct, annotations.Annotations, error) {
+	i := int(f.pollCalls.Add(1)) - 1
+	if i >= len(f.polls) {
+		// Polling past the script (or a case that should never poll) fails
+		// loudly instead of hanging the loop.
+		return v2.BatonActionStatus_BATON_ACTION_STATUS_UNSPECIFIED, "", nil, nil, fmt.Errorf("unexpected status poll %d", i+1)
+	}
+	p := f.polls[i]
+	if p.err != nil {
+		return v2.BatonActionStatus_BATON_ACTION_STATUS_UNKNOWN, "", nil, nil, p.err
+	}
+	if p.resp != nil {
+		return p.status, "scripted", p.resp, nil, nil
+	}
+	return p.status, "scripted", f.invokeRv, nil, nil
+}
+
+// Every seam and poll outcome the wrapper distinguishes: terminal statuses at
+// the invoke seam resolve like terminal polls (in-band FAILED must not become
+// a success), never-populated shapes keep the legacy pass-through, and the
+// poll loop tolerates transient lookup errors and indeterminate statuses up
+// to the shared threshold before failing closed.
+func TestRegisterLegacyActionSeamAndPollOutcomes(t *testing.T) {
+	const (
+		unspecified = v2.BatonActionStatus_BATON_ACTION_STATUS_UNSPECIFIED
+		unknown     = v2.BatonActionStatus_BATON_ACTION_STATUS_UNKNOWN
+		running     = v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING
+		complete    = v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE
+		failed      = v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED
+	)
+	lookupErr := scriptedPollResult{err: context.DeadlineExceeded}
+
+	cases := []struct {
+		name         string
+		invokeID     string
+		invokeStatus v2.BatonActionStatus
+		polls        []scriptedPollResult
+		wantStatus   v2.BatonActionStatus
+		wantErrIn    string
+	}{
+		{"in-band failure at the invoke seam fails", "id-1", failed, nil, failed, "failed"},
+		{"indeterminate seam status polls to an outcome", "id-1", unknown,
+			[]scriptedPollResult{{status: complete}}, complete, ""},
+		{"indeterminate seam status without an id resolves as before", "", unknown, nil, complete, ""},
+		{"unspecified status resolves as before", "id-1", unspecified, nil, complete, ""},
+		{"in-flight without an id resolves as before", "", running, nil, complete, ""},
+		{"threshold consecutive lookup errors fail closed", "id-1", running,
+			[]scriptedPollResult{lookupErr, lookupErr, lookupErr}, failed, "status lookup failed"},
+		{"lookup errors under the threshold recover", "id-1", running,
+			[]scriptedPollResult{lookupErr, lookupErr, {status: complete}}, complete, ""},
+		{"indeterminate polls under the threshold recover", "id-1", running,
+			[]scriptedPollResult{{status: unspecified}, {status: running}, {status: complete}}, complete, ""},
+		{"persistent indeterminate polls fail closed", "id-1", running,
+			[]scriptedPollResult{{status: unspecified}, {status: unspecified}, {status: unspecified}}, failed, "unexpected status"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := t.Context()
+			rv, err := structpb.NewStruct(map[string]any{"done": true})
+			require.NoError(t, err)
+			legacy := &scriptedLegacyActionManager{
+				schema:       v2.BatonActionSchema_builder{Name: "scripted_action"}.Build(),
+				invokeID:     tc.invokeID,
+				invokeStatus: tc.invokeStatus,
+				invokeRv:     rv,
+				polls:        tc.polls,
+			}
+			outer := actions.NewActionManager(ctx)
+			require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy, shortPollIntervals))
+
+			_, st, gotRv, _, err := outer.InvokeAction(ctx, "scripted_action", "", &structpb.Struct{})
+			require.NoError(t, err)
+			require.Equal(t, tc.wantStatus, st)
+			if tc.wantErrIn != "" {
+				require.Contains(t, gotRv.Fields["error"].GetStringValue(), tc.wantErrIn)
+			} else {
+				require.Nil(t, gotRv.Fields["error"])
+			}
+		})
+	}
+}
+
+// The poll loop's warnings must reach the caller's logger: the detached
+// handler context carries it across the goroutine boundary.
+func TestLegacyPollWarningsReachCallerLogger(t *testing.T) {
+	core, observed := observer.New(zap.WarnLevel)
+	ctx := ctxzap.ToContext(t.Context(), zap.New(core))
+
+	rv, err := structpb.NewStruct(map[string]any{"done": true})
+	require.NoError(t, err)
+	legacy := &scriptedLegacyActionManager{
+		schema:       v2.BatonActionSchema_builder{Name: "warned_action"}.Build(),
+		invokeID:     "id-1",
+		invokeStatus: v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING,
+		invokeRv:     rv,
+		polls: []scriptedPollResult{
+			{err: context.DeadlineExceeded},
+			{status: v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE},
+		},
+	}
+	outer := actions.NewActionManager(ctx)
+	require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy, shortPollIntervals))
+
+	_, st, _, _, err := outer.InvokeAction(ctx, "warned_action", "", &structpb.Struct{})
+	require.NoError(t, err)
+	require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE, st)
+	require.Eventually(t, func() bool {
+		return observed.FilterMessage("legacy action status check failed, retrying").Len() == 1
+	}, time.Second, 10*time.Millisecond)
+}
+
+type capturingRegistry struct {
+	handler actions.ActionHandler
+}
+
+func (c *capturingRegistry) Register(_ context.Context, _ *v2.BatonActionSchema, handler actions.ActionHandler) error {
+	c.handler = handler
+	return nil
+}
+
+func (c *capturingRegistry) RegisterAction(_ context.Context, _ string, _ *v2.BatonActionSchema, handler actions.ActionHandler) error {
+	c.handler = handler
+	return nil
+}
+
+// The poll loop's context exit must surface the handler budget's cause, not
+// a bare context error.
+func TestLegacyPollSurfacesHandlerBudgetCause(t *testing.T) {
+	legacy := &scriptedLegacyActionManager{
+		schema:       v2.BatonActionSchema_builder{Name: "budgeted_action"}.Build(),
+		invokeID:     "id-1",
+		invokeStatus: v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING,
+		polls:        slices.Repeat([]scriptedPollResult{{status: v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING}}, 64),
+	}
+	reg := &capturingRegistry{}
+	require.NoError(t, registerLegacyAction(t.Context(), reg, legacy.schema, legacy, shortPollIntervals))
+
+	cause := errors.New("action handler timed out")
+	handlerCtx, cancel := context.WithTimeoutCause(t.Context(), 25*time.Millisecond, cause)
+	defer cancel()
+
+	_, _, err := reg.handler(handlerCtx, &structpb.Struct{})
+	require.ErrorIs(t, err, cause)
+	require.ErrorContains(t, err, "did not reach a terminal status")
+}
+
+// Removing the meaningful-payload guard in the poll loop must fail this
+// test: an indeterminate poll's payload must not displace the invoke
+// response that the fail-closed exit returns.
+func TestLegacyIndeterminatePollPayloadDoesNotDisplaceResponse(t *testing.T) {
+	ctx := t.Context()
+
+	invokePayload, err := structpb.NewStruct(map[string]any{"from": "invoke"})
+	require.NoError(t, err)
+	pollPayload, err := structpb.NewStruct(map[string]any{"from": "poll"})
+	require.NoError(t, err)
+
+	indeterminate := scriptedPollResult{
+		status: v2.BatonActionStatus_BATON_ACTION_STATUS_UNSPECIFIED,
+		resp:   pollPayload,
+	}
+	legacy := &scriptedLegacyActionManager{
+		schema:       v2.BatonActionSchema_builder{Name: "displacing_action"}.Build(),
+		invokeID:     "id-1",
+		invokeStatus: v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING,
+		invokeRv:     invokePayload,
+		polls:        []scriptedPollResult{indeterminate, indeterminate, indeterminate},
+	}
+	outer := actions.NewActionManager(ctx)
+	require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy, shortPollIntervals))
+
+	_, st, gotRv, _, err := outer.InvokeAction(ctx, "displacing_action", "", &structpb.Struct{})
+	require.NoError(t, err)
+	require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED, st)
+	require.Equal(t, "invoke", gotRv.Fields["from"].GetStringValue())
+	require.Contains(t, gotRv.Fields["error"].GetStringValue(), "unexpected status")
+}
+
+// Weakening either half of the interval fallback must fail a case here: the
+// fallback replaces the whole pacing struct, so even a non-default initial
+// paired with a zero cap (which would busy-loop from the second tick) must
+// be re-paced to the defaults' initial tick, which is what the elapsed
+// lower bound proves.
+func TestNonPositivePollIntervalsFallBackToDefaults(t *testing.T) {
+	cases := []struct {
+		name      string
+		intervals legacyPollIntervals
+	}{
+		{"both zero", legacyPollIntervals{}},
+		{"zero cap with a live initial", legacyPollIntervals{initial: 50 * time.Millisecond}},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			rv, err := structpb.NewStruct(map[string]any{"done": true})
+			require.NoError(t, err)
+			legacy := &scriptedLegacyActionManager{
+				schema:       v2.BatonActionSchema_builder{Name: "unpaced_action"}.Build(),
+				invokeID:     "id-1",
+				invokeStatus: v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING,
+				invokeRv:     rv,
+				polls:        []scriptedPollResult{{status: v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE}},
+			}
+			reg := &capturingRegistry{}
+			require.NoError(t, registerLegacyAction(t.Context(), reg, legacy.schema, legacy, tc.intervals))
+
+			start := time.Now()
+			_, _, err = reg.handler(t.Context(), &structpb.Struct{})
+			elapsed := time.Since(start)
+
+			require.NoError(t, err)
+			require.EqualValues(t, 1, legacy.pollCalls.Load())
+			require.GreaterOrEqual(t, elapsed, 900*time.Millisecond)
+		})
+	}
+}
+
+// The inner manager's reported error must survive into the outer error,
+// which replaces the response's error field: at the seam and from a poll
+// alike, the failed outcome carries the inner message, not just the
+// generic wrapper text.
+func TestLegacyFailureCarriesInnerError(t *testing.T) {
+	failedRv, err := structpb.NewStruct(map[string]any{"error": "boom from inner"})
+	require.NoError(t, err)
+
+	t.Run("at the invoke seam", func(t *testing.T) {
+		ctx := t.Context()
+		legacy := &scriptedLegacyActionManager{
+			schema:       v2.BatonActionSchema_builder{Name: "inner_error_action"}.Build(),
+			invokeID:     "id-1",
+			invokeStatus: v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED,
+			invokeRv:     failedRv,
+		}
+		outer := actions.NewActionManager(ctx)
+		require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy, shortPollIntervals))
+
+		_, st, gotRv, _, err := outer.InvokeAction(ctx, "inner_error_action", "", &structpb.Struct{})
+		require.NoError(t, err)
+		require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED, st)
+		require.Contains(t, gotRv.Fields["error"].GetStringValue(), "boom from inner")
+	})
+
+	t.Run("from a poll", func(t *testing.T) {
+		ctx := t.Context()
+		legacy := &scriptedLegacyActionManager{
+			schema:       v2.BatonActionSchema_builder{Name: "inner_error_action"}.Build(),
+			invokeID:     "id-1",
+			invokeStatus: v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING,
+			polls: []scriptedPollResult{
+				{status: v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED, resp: failedRv},
+			},
+		}
+		outer := actions.NewActionManager(ctx)
+		require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy, shortPollIntervals))
+
+		_, st, gotRv, _, err := outer.InvokeAction(ctx, "inner_error_action", "", &structpb.Struct{})
+		require.NoError(t, err)
+		require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED, st)
+		require.Contains(t, gotRv.Fields["error"].GetStringValue(), "boom from inner")
+	})
+}
+
+// A stale in-flight payload must not be reported as the failure cause when
+// the settling poll itself carries no payload.
+func TestLegacyStaleInFlightPayloadIsNotTheFailureCause(t *testing.T) {
+	ctx := t.Context()
+
+	staleRv, err := structpb.NewStruct(map[string]any{"error": "poll1-snapshot"})
+	require.NoError(t, err)
+	legacy := &scriptedLegacyActionManager{
+		schema:       v2.BatonActionSchema_builder{Name: "stale_error_action"}.Build(),
+		invokeID:     "id-1",
+		invokeStatus: v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING,
+		polls: []scriptedPollResult{
+			{status: v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING, resp: staleRv},
+			{status: v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED},
+		},
+	}
+	outer := actions.NewActionManager(ctx)
+	require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy, shortPollIntervals))
+
+	_, st, gotRv, _, err := outer.InvokeAction(ctx, "stale_error_action", "", &structpb.Struct{})
+	require.NoError(t, err)
+	require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED, st)
+	require.NotContains(t, gotRv.Fields["error"].GetStringValue(), "poll1-snapshot")
+	require.Contains(t, gotRv.Fields["error"].GetStringValue(), "failed")
+}
diff --git a/vendor/go.uber.org/zap/zaptest/observer/logged_entry.go b/vendor/go.uber.org/zap/zaptest/observer/logged_entry.go
new file mode 100644
index 000000000..ef89e25c3
--- /dev/null
+++ b/vendor/go.uber.org/zap/zaptest/observer/logged_entry.go
@@ -0,0 +1,39 @@
+// Copyright (c) 2017 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package observer
+
+import "go.uber.org/zap/zapcore"
+
+// A LoggedEntry is an encoding-agnostic representation of a log message.
+// Field availability is context dependent.
+type LoggedEntry struct {
+	zapcore.Entry
+	Context []zapcore.Field
+}
+
+// ContextMap returns a map for all fields in Context.
+func (e LoggedEntry) ContextMap() map[string]interface{} {
+	encoder := zapcore.NewMapObjectEncoder()
+	for _, f := range e.Context {
+		f.AddTo(encoder)
+	}
+	return encoder.Fields
+}
diff --git a/vendor/go.uber.org/zap/zaptest/observer/observer.go b/vendor/go.uber.org/zap/zaptest/observer/observer.go
new file mode 100644
index 000000000..4f7ce0ec6
--- /dev/null
+++ b/vendor/go.uber.org/zap/zaptest/observer/observer.go
@@ -0,0 +1,203 @@
+// Copyright (c) 2016-2022 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+// Package observer provides a zapcore.Core that keeps an in-memory,
+// encoding-agnostic representation of log entries. It's useful for
+// applications that want to unit test their log output without tying their
+// tests to a particular output encoding.
+package observer // import "go.uber.org/zap/zaptest/observer"
+
+import (
+	"strings"
+	"sync"
+	"time"
+
+	"go.uber.org/zap/internal"
+	"go.uber.org/zap/zapcore"
+)
+
+// ObservedLogs is a concurrency-safe, ordered collection of observed logs.
+type ObservedLogs struct {
+	mu   sync.RWMutex
+	logs []LoggedEntry
+}
+
+// Len returns the number of items in the collection.
+func (o *ObservedLogs) Len() int {
+	o.mu.RLock()
+	n := len(o.logs)
+	o.mu.RUnlock()
+	return n
+}
+
+// All returns a copy of all the observed logs.
+func (o *ObservedLogs) All() []LoggedEntry {
+	o.mu.RLock()
+	ret := make([]LoggedEntry, len(o.logs))
+	copy(ret, o.logs)
+	o.mu.RUnlock()
+	return ret
+}
+
+// TakeAll returns a copy of all the observed logs, and truncates the observed
+// slice.
+func (o *ObservedLogs) TakeAll() []LoggedEntry {
+	o.mu.Lock()
+	ret := o.logs
+	o.logs = nil
+	o.mu.Unlock()
+	return ret
+}
+
+// AllUntimed returns a copy of all the observed logs, but overwrites the
+// observed timestamps with time.Time's zero value. This is useful when making
+// assertions in tests.
+func (o *ObservedLogs) AllUntimed() []LoggedEntry {
+	ret := o.All()
+	for i := range ret {
+		ret[i].Time = time.Time{}
+	}
+	return ret
+}
+
+// FilterLevelExact filters entries to those logged at exactly the given level.
+func (o *ObservedLogs) FilterLevelExact(level zapcore.Level) *ObservedLogs {
+	return o.Filter(func(e LoggedEntry) bool {
+		return e.Level == level
+	})
+}
+
+// FilterMessage filters entries to those that have the specified message.
+func (o *ObservedLogs) FilterMessage(msg string) *ObservedLogs {
+	return o.Filter(func(e LoggedEntry) bool {
+		return e.Message == msg
+	})
+}
+
+// FilterLoggerName filters entries to those logged through logger with the specified logger name.
+func (o *ObservedLogs) FilterLoggerName(name string) *ObservedLogs {
+	return o.Filter(func(e LoggedEntry) bool {
+		return e.LoggerName == name
+	})
+}
+
+// FilterMessageSnippet filters entries to those that have a message containing the specified snippet.
+func (o *ObservedLogs) FilterMessageSnippet(snippet string) *ObservedLogs {
+	return o.Filter(func(e LoggedEntry) bool {
+		return strings.Contains(e.Message, snippet)
+	})
+}
+
+// FilterField filters entries to those that have the specified field.
+func (o *ObservedLogs) FilterField(field zapcore.Field) *ObservedLogs {
+	return o.Filter(func(e LoggedEntry) bool {
+		for _, ctxField := range e.Context {
+			if ctxField.Equals(field) {
+				return true
+			}
+		}
+		return false
+	})
+}
+
+// FilterFieldKey filters entries to those that have the specified key.
+func (o *ObservedLogs) FilterFieldKey(key string) *ObservedLogs {
+	return o.Filter(func(e LoggedEntry) bool {
+		for _, ctxField := range e.Context {
+			if ctxField.Key == key {
+				return true
+			}
+		}
+		return false
+	})
+}
+
+// Filter returns a copy of this ObservedLogs containing only those entries
+// for which the provided function returns true.
+func (o *ObservedLogs) Filter(keep func(LoggedEntry) bool) *ObservedLogs {
+	o.mu.RLock()
+	defer o.mu.RUnlock()
+
+	var filtered []LoggedEntry
+	for _, entry := range o.logs {
+		if keep(entry) {
+			filtered = append(filtered, entry)
+		}
+	}
+	return &ObservedLogs{logs: filtered}
+}
+
+func (o *ObservedLogs) add(log LoggedEntry) {
+	o.mu.Lock()
+	o.logs = append(o.logs, log)
+	o.mu.Unlock()
+}
+
+// New creates a new Core that buffers logs in memory (without any encoding).
+// It's particularly useful in tests.
+func New(enab zapcore.LevelEnabler) (zapcore.Core, *ObservedLogs) {
+	ol := &ObservedLogs{}
+	return &contextObserver{
+		LevelEnabler: enab,
+		logs:         ol,
+	}, ol
+}
+
+type contextObserver struct {
+	zapcore.LevelEnabler
+	logs    *ObservedLogs
+	context []zapcore.Field
+}
+
+var (
+	_ zapcore.Core            = (*contextObserver)(nil)
+	_ internal.LeveledEnabler = (*contextObserver)(nil)
+)
+
+func (co *contextObserver) Level() zapcore.Level {
+	return zapcore.LevelOf(co.LevelEnabler)
+}
+
+func (co *contextObserver) Check(ent zapcore.Entry, ce *zapcore.CheckedEntry) *zapcore.CheckedEntry {
+	if co.Enabled(ent.Level) {
+		return ce.AddCore(ent, co)
+	}
+	return ce
+}
+
+func (co *contextObserver) With(fields []zapcore.Field) zapcore.Core {
+	return &contextObserver{
+		LevelEnabler: co.LevelEnabler,
+		logs:         co.logs,
+		context:      append(co.context[:len(co.context):len(co.context)], fields...),
+	}
+}
+
+func (co *contextObserver) Write(ent zapcore.Entry, fields []zapcore.Field) error {
+	all := make([]zapcore.Field, 0, len(fields)+len(co.context))
+	all = append(all, co.context...)
+	all = append(all, fields...)
+	co.logs.add(LoggedEntry{ent, all})
+	return nil
+}
+
+func (co *contextObserver) Sync() error {
+	return nil
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index a678f105c..ac3ac554d 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -964,6 +964,7 @@ go.uber.org/zap/internal/exit
 go.uber.org/zap/internal/pool
 go.uber.org/zap/internal/stacktrace
 go.uber.org/zap/zapcore
+go.uber.org/zap/zaptest/observer
 # go.yaml.in/yaml/v3 v3.0.4
 ## explicit; go 1.16
 go.yaml.in/yaml/v3