Merge pull request #709 from DingTalk-Real-AI/codex/changelog-v1.0.53-beta.5

docs(changelog): seal v1.0.53-beta.5 notes

CHANGELOG.md

@@ -6,6 +6,10 @@ The format is inspired by [Keep a Changelog](https://keepachangelog.com/) and th
 
 ## [Unreleased]
 
+## [1.0.53-beta.5] - 2026-07-21
+
+This beta validates long-running access-token recovery and the faster, recoverable guarded release path introduced after v1.0.53-beta.4.
+
 ### Changed
 
 - **Fast guarded beta and stable releases** — successful local release checks now leave a six-hour proof bound to the exact version, commit, repository identity, remote `main`, and stable baseline, so the subsequent guarded `--publish` invocation revalidates authority without repeating tests and packaging. A default-branch governance smoke uses the same dedicated immutable-release credential as the tag workflow before any tag is allocated.
@@ -13,6 +17,7 @@ The format is inspired by [Keep a Changelog](https://keepachangelog.com/) and th
 
 ### Fixed
 
+- **Long-running event authentication recovery** — personal and portal event streams resolve the current access token for every ticket request, refresh a server-rejected token with compare-and-refresh semantics, and reconnect with backoff when refresh is temporarily blocked by network failures, rate limits, or 5xx responses.
 - **Consistent access-token caching and errors** — runtime, recovery, Skill, PAT polling, and personal/portal event clients now resolve user access tokens through one expiry- and publication-aware manager, so long-running processes reload rotated credentials while keychain, refresh, parse, permission, and cancellation failures remain observable instead of being collapsed into “not authenticated.”
 - **Tag-push GitHub Release publication** — Draft publication now locks one GitHub Release database ID, verifies its exact tag, channel, notes, recovery marker, asset set, and uploaded bytes, then publishes and rechecks that same ID as immutable. Recovery runs use the trusted default-branch release helpers instead of the sealed tag's historical scripts, fixing the Draft-only `GET /releases/tags/{tag}` 404 without allowing the release identity to drift during recovery.
 - **Release preflight reliability** — source-mode installer tests now use isolated temporary checkouts and HOME directories instead of overwriting and deleting the real repository `dws` binary, release preflight explicitly rebuilds before policy checks, and the full-suite runner gives the growing script package a non-flaky five-minute per-suite budget.

internal/app/event_command.go

@@ -56,6 +56,7 @@ var (
 	eventNewEventSource        = newEventSource
 	eventNewDingtalkSource     = source.New
 	eventResolveAccessToken    = ResolveAuxiliaryAccessToken
+	eventForceRefreshRejected  = forceRefreshRejectedAccessToken
 	eventBusRun                = bus.Run
 	eventReadyFDFromEnv        = busctl.ReadyFDFromEnv
 	eventResolvePersonal       = resolvePersonalEventIdentity
@@ -436,6 +437,9 @@ func newEventSource(_ context.Context, configDir, clientID, clientSecret string,
 			AccessTokenProvider: func(ctx context.Context) (string, error) {
 				return eventResolveAccessToken(ctx, configDir, "")
 			},
+			ForceRefreshToken: func(ctx context.Context, rejectedToken string) (string, error) {
+				return eventForceRefreshRejected(ctx, configDir, rejectedToken)
+			},
 			SourceID:     eventStreamSourceID(streamOpts.SourceID),
 			Mode:         streamOpts.Mode,
 			ClientID:     portalClientID,

internal/app/event_force_refresh_wire_test.go

@@ -0,0 +1,47 @@
+package app
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/source"
+)
+
+// TestCrossPlatformCoverageNewEventSourceWiresForceRefreshRejectedToken asserts the portal ticket
+// source receives a ForceRefreshToken callback that forwards the actual
+// rejected token into the app-level compare-and-refresh chain.
+func TestCrossPlatformCoverageNewEventSourceWiresForceRefreshRejectedToken(t *testing.T) {
+	oldNew, oldRefresh := eventNewDingtalkSource, eventForceRefreshRejected
+	t.Cleanup(func() { eventNewDingtalkSource, eventForceRefreshRejected = oldNew, oldRefresh })
+
+	var captured source.Config
+	eventNewDingtalkSource = func(cfg source.Config, _ ...source.SourceOption) (*source.DingtalkSource, error) {
+		captured = cfg
+		return &source.DingtalkSource{}, nil
+	}
+	var gotDir, gotRejected string
+	eventForceRefreshRejected = func(_ context.Context, configDir, rejectedToken string) (string, error) {
+		gotDir, gotRejected = configDir, rejectedToken
+		return "fresh", nil
+	}
+	if _, err := newEventSource(context.Background(), "config-dir", "client", "secret", eventStreamTicketOptions{Mode: "custom"}); err != nil {
+		t.Fatal(err)
+	}
+	if captured.PortalTicket == nil || captured.PortalTicket.ForceRefreshToken == nil {
+		t.Fatal("ForceRefreshToken not wired into portal ticket config")
+	}
+	tok, err := captured.PortalTicket.ForceRefreshToken(context.Background(), "rejected-token")
+	if err != nil || tok != "fresh" {
+		t.Fatalf("force refresh = %q, %v", tok, err)
+	}
+	if gotDir != "config-dir" || gotRejected != "rejected-token" {
+		t.Fatalf("wiring passed dir %q rejected %q", gotDir, gotRejected)
+	}
+
+	fail := errors.New("refresh failed")
+	eventForceRefreshRejected = func(context.Context, string, string) (string, error) { return "", fail }
+	if _, err := captured.PortalTicket.ForceRefreshToken(context.Background(), "x"); !errors.Is(err, fail) {
+		t.Fatalf("refresh error = %v", err)
+	}
+}

internal/app/event_personal_command.go

@@ -131,6 +131,7 @@ var (
 	personalFindProcess                 = os.FindProcess
 	personalSignalProcess               = (*os.Process).Signal
 	personalResolveAuxiliaryAccessToken = ResolveAuxiliaryAccessToken
+	personalForceRefreshRejectedToken   = forceRefreshRejectedAccessToken
 	personalLoadTokenData               = authpkg.LoadTokenData
 	personalClientID                    = authpkg.ClientID
 	personalResolveAppCredentialsStrict = authpkg.ResolveAppCredentialsStrict
@@ -832,6 +833,9 @@ func newPersonalStreamSource(ctx context.Context, opts personalStreamSourceOptio
 		AccessTokenProvider: func(ctx context.Context) (string, error) {
 			return personalResolveAuxiliaryAccessToken(ctx, opts.ConfigDir, "")
 		},
+		ForceRefreshToken: func(ctx context.Context, rejectedToken string) (string, error) {
+			return personalForceRefreshRejectedToken(ctx, opts.ConfigDir, rejectedToken)
+		},
 		ClientID:     clientID,
 		ClientSecret: clientSecret,
 		SourceID:     opts.Identity.SourceID,

internal/app/event_personal_force_refresh_wire_test.go

@@ -0,0 +1,54 @@
+package app
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	dwsevent "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/personal"
+)
+
+// TestCrossPlatformCoverageNewPersonalStreamSourceWiresForceRefreshRejectedToken asserts the
+// personal stream source receives a ForceRefreshToken callback that forwards
+// the rejected token into the app-level compare-and-refresh chain.
+func TestCrossPlatformCoverageNewPersonalStreamSourceWiresForceRefreshRejectedToken(t *testing.T) {
+	oldAux := personalResolveAuxiliaryAccessToken
+	oldRefresh := personalForceRefreshRejectedToken
+	t.Cleanup(func() {
+		personalResolveAuxiliaryAccessToken = oldAux
+		personalForceRefreshRejectedToken = oldRefresh
+	})
+	personalResolveAuxiliaryAccessToken = func(context.Context, string, string) (string, error) {
+		return "old-token", nil
+	}
+	refreshErr := errors.New("refresh rejected")
+	var gotDir, gotRejected string
+	personalForceRefreshRejectedToken = func(_ context.Context, configDir, rejectedToken string) (string, error) {
+		gotDir, gotRejected = configDir, rejectedToken
+		return "", refreshErr
+	}
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusUnauthorized)
+	}))
+	defer srv.Close()
+
+	src, err := newPersonalStreamSource(context.Background(), personalStreamSourceOptions{
+		ConfigDir: "config-dir",
+		Identity:  personal.Identity{ClientID: "client", SourceID: "source"},
+		TicketURL: srv.URL,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	// The 401 ticket response routes the rejected token through the wired
+	// ForceRefreshToken; the unknown refresh failure stays fatal.
+	if err := src.Start(context.Background(), func(*dwsevent.RawEvent) {}); !errors.Is(err, refreshErr) {
+		t.Fatalf("Start() error = %v, want wrapped refresh error", err)
+	}
+	if gotDir != "config-dir" || gotRejected != "old-token" {
+		t.Fatalf("refresh wiring got dir %q rejected %q", gotDir, gotRejected)
+	}
+}

internal/auth/device_flow.go

@@ -722,6 +722,6 @@ func isInvalidGrantError(err error) bool {
 	if err == nil {
 		return false
 	}
-	msg := strings.ToLower(err.Error())
+	msg := strings.ToLower(err.Error() + " " + httpStatusResponseBody(err))
 	return strings.Contains(msg, "invalid_grant") || (strings.Contains(msg, "code") && strings.Contains(msg, "expired"))
 }

internal/auth/oauth_helpers.go

@@ -237,12 +237,21 @@ func (p *OAuthProvider) postJSON(ctx context.Context, endpoint string, body any)
 	}
 	defer resp.Body.Close()
 
-	data, err := io.ReadAll(io.LimitReader(resp.Body, config.MaxResponseBodySize))
-	if err != nil {
-		return nil, fmt.Errorf("reading response: %w", err)
-	}
+	data, readErr := io.ReadAll(io.LimitReader(resp.Body, config.MaxResponseBodySize))
 	if resp.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("HTTP %d: %s", resp.StatusCode, truncateBody(data, 200))
+		// Preserve structured HTTP status semantics even when the response
+		// body is truncated. The body is diagnostic-only here, so read it
+		// best-effort and classify retryability from the status code.
+		if readErr != nil {
+			data = nil
+		}
+		return nil, &HTTPStatusError{
+			StatusCode:   resp.StatusCode,
+			responseBody: truncateBody(data, 200),
+		}
+	}
+	if readErr != nil {
+		return nil, fmt.Errorf("reading response: %w", readErr)
 	}
 	return data, nil
 }

internal/auth/oauth_postjson_test.go

@@ -0,0 +1,69 @@
+// Copyright 2026 Alibaba Group
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package auth
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"testing"
+)
+
+type postJSONRoundTripFunc func(*http.Request) (*http.Response, error)
+
+func (f postJSONRoundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) {
+	return f(req)
+}
+
+type oauthBrokenBody struct{}
+
+func (oauthBrokenBody) Read([]byte) (int, error) { return 0, io.ErrUnexpectedEOF }
+func (oauthBrokenBody) Close() error             { return nil }
+
+func TestCrossPlatformCoveragePostJSONTruncatedErrorBodyKeepsHTTPStatus(t *testing.T) {
+	for _, status := range []int{http.StatusTooManyRequests, http.StatusServiceUnavailable} {
+		t.Run(http.StatusText(status), func(t *testing.T) {
+			provider := &OAuthProvider{httpClient: &http.Client{Transport: postJSONRoundTripFunc(func(*http.Request) (*http.Response, error) {
+				return &http.Response{StatusCode: status, Body: oauthBrokenBody{}, Header: make(http.Header)}, nil
+			})}}
+
+			_, err := provider.postJSON(context.Background(), "https://oauth.test/token", map[string]string{"grantType": "refresh_token"})
+			var statusErr *HTTPStatusError
+			if !errors.As(err, &statusErr) || statusErr.StatusCode != status {
+				t.Fatalf("postJSON() error = %v, want HTTPStatusError %d", err, status)
+			}
+			if errors.Is(err, io.ErrUnexpectedEOF) {
+				t.Fatalf("HTTP status error should not expose diagnostic body read failure: %v", err)
+			}
+			if got := ClassifyRefreshFailure(err); got != RefreshFailureTransient {
+				t.Fatalf("ClassifyRefreshFailure() = %s, want transient", got)
+			}
+		})
+	}
+}
+
+func TestCrossPlatformCoveragePostJSONOKTruncatedBodyIsTransient(t *testing.T) {
+	provider := &OAuthProvider{httpClient: &http.Client{Transport: postJSONRoundTripFunc(func(*http.Request) (*http.Response, error) {
+		return &http.Response{StatusCode: http.StatusOK, Body: oauthBrokenBody{}, Header: make(http.Header)}, nil
+	})}}
+
+	_, err := provider.postJSON(context.Background(), "https://oauth.test/token", map[string]string{"grantType": "refresh_token"})
+	if !errors.Is(err, io.ErrUnexpectedEOF) {
+		t.Fatalf("postJSON() error = %v, want io.ErrUnexpectedEOF", err)
+	}
+	if got := ClassifyRefreshFailure(err); got != RefreshFailureTransient {
+		t.Fatalf("ClassifyRefreshFailure() = %s, want transient", got)
+	}
+}

internal/auth/oauth_provider.go

@@ -18,6 +18,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"html"
 	"io"
 	"log/slog"
 	"net"
@@ -301,7 +302,7 @@ func (p *OAuthProvider) Login(ctx context.Context, force bool) (*TokenData, erro
 			callbackTokenMu.Unlock()
 
 			w.Header().Set("Content-Type", "text/html; charset=utf-8")
-			_, _ = fmt.Fprintf(w, "<html><body><h1>授权失败</h1><p>%s</p></body></html>", exchangeErr.Error())
+			_, _ = fmt.Fprintf(w, "<html><body><h1>授权失败</h1><p>%s</p></body></html>", html.EscapeString(oauthExchangeDisplayError(exchangeErr)))
 			select {
 			case resultCh <- callbackResult{err: exchangeErr}:
 			default:
@@ -642,6 +643,14 @@ continueLogin:
 	return tokenData, nil
 }
 
+func oauthExchangeDisplayError(err error) string {
+	var statusErr *HTTPStatusError
+	if errors.As(err, &statusErr) && statusErr != nil {
+		return fmt.Sprintf("HTTP %d: token exchange failed", statusErr.StatusCode)
+	}
+	return err.Error()
+}
+
 // GetTokenSnapshot returns a valid token together with its expiry metadata.
 // Storage and refresh failures retain their original cause; only a confirmed
 // missing credential is reported as ErrTokenDataNotFound.
@@ -665,7 +674,12 @@ func (p *OAuthProvider) GetTokenSnapshot(ctx context.Context) (*TokenData, error
 		if rErr == nil {
 			return refreshed, nil
 		}
-		_ = oauthMarkProfile(p.configDir, TokenProfileSelector(data), ProfileStatusExpired)
+		// A network, timeout, rate-limit or 5xx failure does not invalidate the
+		// refresh credential. Keep the profile active so a long-running source
+		// can retry after backoff. Terminal and unknown failures remain fatal.
+		if ClassifyRefreshFailure(rErr) != RefreshFailureTransient {
+			_ = oauthMarkProfile(p.configDir, TokenProfileSelector(data), ProfileStatusExpired)
+		}
 		if p.logger != nil {
 			p.logger.Warn(i18n.T("refresh_token 刷新失败"), "error", rErr)
 		}

internal/auth/refresh_failure.go

@@ -0,0 +1,89 @@
+// Copyright 2026 Alibaba Group
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package auth
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+)
+
+// RefreshFailureClass separates refresh failures that may recover after a
+// delay from failures that require new credentials or local intervention.
+type RefreshFailureClass string
+
+const (
+	RefreshFailureUnknown   RefreshFailureClass = "unknown"
+	RefreshFailureTransient RefreshFailureClass = "transient"
+	RefreshFailureTerminal  RefreshFailureClass = "terminal"
+)
+
+// HTTPStatusError preserves an OAuth endpoint status for structured retry
+// decisions without copying an untrusted response body into logs.
+type HTTPStatusError struct {
+	StatusCode   int
+	responseBody string
+}
+
+func (e *HTTPStatusError) Error() string {
+	if e == nil {
+		return "OAuth endpoint request failed"
+	}
+	return fmt.Sprintf("HTTP %d", e.StatusCode)
+}
+
+func httpStatusResponseBody(err error) string {
+	var statusErr *HTTPStatusError
+	if !errors.As(err, &statusErr) || statusErr == nil {
+		return ""
+	}
+	return statusErr.responseBody
+}
+
+// ClassifyRefreshFailure uses only structured transport and HTTP signals.
+// Unknown errors, including parse, keychain and persistence failures, remain
+// fatal so a long-running source cannot retry an error that needs user action.
+func ClassifyRefreshFailure(err error) RefreshFailureClass {
+	if err == nil {
+		return RefreshFailureUnknown
+	}
+	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+		return RefreshFailureTransient
+	}
+	if errors.Is(err, io.ErrUnexpectedEOF) {
+		return RefreshFailureTransient
+	}
+	var netErr net.Error
+	if errors.As(err, &netErr) {
+		return RefreshFailureTransient
+	}
+	var statusErr *HTTPStatusError
+	if !errors.As(err, &statusErr) || statusErr == nil {
+		return RefreshFailureUnknown
+	}
+	if statusErr.StatusCode == http.StatusRequestTimeout ||
+		statusErr.StatusCode == http.StatusTooManyRequests ||
+		statusErr.StatusCode >= http.StatusInternalServerError {
+		return RefreshFailureTransient
+	}
+	if statusErr.StatusCode == http.StatusBadRequest ||
+		statusErr.StatusCode == http.StatusUnauthorized ||
+		statusErr.StatusCode == http.StatusForbidden {
+		return RefreshFailureTerminal
+	}
+	return RefreshFailureUnknown
+}

internal/auth/refresh_failure_test.go

@@ -0,0 +1,155 @@
+// Copyright 2026 Alibaba Group
+// Licensed under the Apache License, Version 2.0 (the "License");
+
+package auth
+
+import (
+	"context"
+	"errors"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/pkg/edition"
+)
+
+func TestCrossPlatformCoverageClassifyRefreshFailureUsesStructuredSignals(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want RefreshFailureClass
+	}{
+		{name: "deadline", err: context.DeadlineExceeded, want: RefreshFailureTransient},
+		{name: "network", err: &url.Error{Op: "Post", URL: "https://oauth.test", Err: context.DeadlineExceeded}, want: RefreshFailureTransient},
+		{name: "request timeout", err: &HTTPStatusError{StatusCode: http.StatusRequestTimeout}, want: RefreshFailureTransient},
+		{name: "rate limited", err: &HTTPStatusError{StatusCode: http.StatusTooManyRequests}, want: RefreshFailureTransient},
+		{name: "server unavailable", err: &HTTPStatusError{StatusCode: http.StatusServiceUnavailable}, want: RefreshFailureTransient},
+		{name: "refresh rejected", err: &HTTPStatusError{StatusCode: http.StatusUnauthorized}, want: RefreshFailureTerminal},
+		{name: "invalid grant", err: &HTTPStatusError{StatusCode: http.StatusBadRequest}, want: RefreshFailureTerminal},
+		{name: "forbidden", err: &HTTPStatusError{StatusCode: http.StatusForbidden}, want: RefreshFailureTerminal},
+		{name: "local persistence", err: errors.New("save refreshed token failed"), want: RefreshFailureUnknown},
+		{name: "nil error", err: nil, want: RefreshFailureUnknown},
+		{name: "dns failure", err: &net.DNSError{Err: "no such host", Name: "oauth.test"}, want: RefreshFailureTransient},
+		{name: "redirect status", err: &HTTPStatusError{StatusCode: http.StatusFound}, want: RefreshFailureUnknown},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := ClassifyRefreshFailure(tt.err); got != tt.want {
+				t.Fatalf("ClassifyRefreshFailure() = %q, want %q", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestCrossPlatformCoverageHTTPStatusErrorRetainsStatusThroughWrapping(t *testing.T) {
+	want := &HTTPStatusError{StatusCode: http.StatusTooManyRequests}
+	err := errors.Join(errors.New("refresh failed"), want)
+	if got := ClassifyRefreshFailure(err); got != RefreshFailureTransient {
+		t.Fatalf("ClassifyRefreshFailure() = %q, want transient", got)
+	}
+	var statusErr *HTTPStatusError
+	if !errors.As(err, &statusErr) || statusErr.StatusCode != http.StatusTooManyRequests {
+		t.Fatalf("HTTP status error not retained: %v", err)
+	}
+	if got, want := statusErr.Error(), "HTTP 429"; got != want {
+		t.Fatalf("HTTP status error = %q, want %q", got, want)
+	}
+	var nilStatus *HTTPStatusError
+	if got, want := nilStatus.Error(), "OAuth endpoint request failed"; got != want {
+		t.Fatalf("nil HTTP status error = %q, want %q", got, want)
+	}
+}
+
+func TestCrossPlatformCoverageOAuthExchangeDisplayErrorFallsBackToPlainError(t *testing.T) {
+	if got, want := oauthExchangeDisplayError(&HTTPStatusError{StatusCode: http.StatusBadGateway}), "HTTP 502: token exchange failed"; got != want {
+		t.Fatalf("status display error = %q, want %q", got, want)
+	}
+	if got, want := oauthExchangeDisplayError(errors.New("exchange failed")), "exchange failed"; got != want {
+		t.Fatalf("plain display error = %q, want %q", got, want)
+	}
+}
+
+func TestCrossPlatformCoveragePostJSONClassifiesStatusWithoutLoggingResponseBody(t *testing.T) {
+	const secretBody = `{"refreshToken":"must-not-reach-logs"}`
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusServiceUnavailable)
+		_, _ = w.Write([]byte(secretBody))
+	}))
+	defer server.Close()
+
+	provider := &OAuthProvider{httpClient: server.Client()}
+	_, err := provider.postJSON(context.Background(), server.URL, map[string]string{"grantType": "refresh_token"})
+	if got := ClassifyRefreshFailure(err); got != RefreshFailureTransient {
+		t.Fatalf("ClassifyRefreshFailure() = %q, want transient: %v", got, err)
+	}
+	if strings.Contains(err.Error(), "must-not-reach-logs") {
+		t.Fatalf("postJSON error leaked response body: %v", err)
+	}
+	if got := httpStatusResponseBody(err); !strings.Contains(got, "must-not-reach-logs") {
+		t.Fatalf("postJSON did not retain bounded response details for internal classification: %q", got)
+	}
+}
+
+func TestCrossPlatformCoverageGetTokenSnapshotOnlyExpiresProfileForNonTransientRefreshFailures(t *testing.T) {
+	oldLoad := oauthLoadToken
+	oldLoadLocked := oauthLoadTokenLocked
+	oldAcquire := oauthAcquireLock
+	oldRefresh := oauthRefreshToken
+	oldMark := oauthMarkProfile
+	oldEdition := edition.Get()
+	t.Cleanup(func() {
+		oauthLoadToken = oldLoad
+		oauthLoadTokenLocked = oldLoadLocked
+		oauthAcquireLock = oldAcquire
+		oauthRefreshToken = oldRefresh
+		oauthMarkProfile = oldMark
+		edition.Override(oldEdition)
+	})
+	edition.Override(&edition.Hooks{})
+
+	expired := &TokenData{
+		AccessToken:  "expired-access",
+		ExpiresAt:    time.Now().Add(-time.Hour),
+		RefreshToken: "refresh",
+		RefreshExpAt: time.Now().Add(time.Hour),
+		CorpID:       "corp",
+		UserID:       "user",
+	}
+	oauthLoadToken = func(string) (*TokenData, error) { return expired, nil }
+	oauthLoadTokenLocked = func(string, string) (*TokenData, error) { return expired, nil }
+	oauthAcquireLock = func(context.Context, string) (*DualLock, error) { return &DualLock{}, nil }
+
+	markCalls := 0
+	oauthMarkProfile = func(_, _, status string) error {
+		if status != ProfileStatusExpired {
+			t.Fatalf("profile status = %q, want %q", status, ProfileStatusExpired)
+		}
+		markCalls++
+		return nil
+	}
+	provider := NewOAuthProvider(t.TempDir(), nil)
+
+	oauthRefreshToken = func(*OAuthProvider, context.Context, *TokenData) (*TokenData, error) {
+		return nil, &HTTPStatusError{StatusCode: http.StatusServiceUnavailable}
+	}
+	if _, err := provider.GetTokenSnapshot(context.Background()); ClassifyRefreshFailure(err) != RefreshFailureTransient {
+		t.Fatalf("transient refresh error = %v", err)
+	}
+	if markCalls != 0 {
+		t.Fatalf("transient refresh marked profile expired %d times", markCalls)
+	}
+
+	oauthRefreshToken = func(*OAuthProvider, context.Context, *TokenData) (*TokenData, error) {
+		return nil, &HTTPStatusError{StatusCode: http.StatusUnauthorized}
+	}
+	if _, err := provider.GetTokenSnapshot(context.Background()); ClassifyRefreshFailure(err) != RefreshFailureTerminal {
+		t.Fatalf("terminal refresh error = %v", err)
+	}
+	if markCalls != 1 {
+		t.Fatalf("terminal refresh marked profile expired %d times, want 1", markCalls)
+	}
+}

internal/event/source/coverage_edges_test.go

@@ -414,7 +414,7 @@ func TestCrossPlatformCoveragePortalStartEndToEndAndFailures(t *testing.T) {
 		return s
 	}
 	network := &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) { return nil, errSourceInjected })}
-	s := makeSource(&PortalTicketConfig{TicketURL: "https://x", AccessToken: "t", SourceID: "s", HTTPClient: network})
+	s := makeSource(&PortalTicketConfig{TicketURL: "https://x", AccessToken: "t", SourceID: "s", HTTPClient: network, DisableReconnect: true})
 	if err := s.Start(context.Background(), func(*dwsevent.RawEvent) {}); err == nil {
 		t.Fatal("ticket failure expected")
 	}
@@ -483,7 +483,7 @@ func TestCrossPlatformCoveragePortalStartHandshakeReadAndAckErrors(t *testing.T)
 	makeSource := func(endpoint string) *DingtalkSource {
 		s, err := New(Config{PortalTicket: &PortalTicketConfig{
 			TicketURL: "https://ticket", AccessToken: "token", SourceID: "source",
-			HTTPClient: staticPersonalTicketClient(endpoint, ""),
+			HTTPClient: staticPersonalTicketClient(endpoint, ""), DisableReconnect: true,
 		}})
 		if err != nil {
 			t.Fatal(err)

internal/event/source/personal.go

@@ -27,6 +27,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	authpkg "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/auth"
 	dwsevent "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event"
 	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/pkg/config"
 	"github.com/gorilla/websocket"
@@ -43,6 +44,7 @@ const (
 type PersonalConfig struct {
 	AccessToken         string
 	AccessTokenProvider AccessTokenProvider
+	ForceRefreshToken   ForceRefreshTokenFn
 	ClientID            string
 	ClientSecret        string
 	SourceID            string
@@ -57,6 +59,13 @@ type PersonalConfig struct {
 
 type AccessTokenProvider func(context.Context) (string, error)
 
+// ForceRefreshTokenFn rotates an access token that the server has just
+// rejected (HTTP 401). It receives the exact rejected token so the caller's
+// compare-and-refresh logic can skip the refresh when another goroutine has
+// already rotated it, and returns the fresh token to retry with. Optional:
+// when nil a 401 stays fatal, matching the previous behavior.
+type ForceRefreshTokenFn func(ctx context.Context, rejectedToken string) (string, error)
+
 type PersonalSource struct {
 	cfg     PersonalConfig
 	machine *Machine
@@ -197,8 +206,29 @@ func (s *PersonalSource) runAttempt(ctx context.Context, emit dwsevent.EmitFn) (
 func (s *PersonalSource) fetchTicket(ctx context.Context) (*ticketResponse, error) {
 	accessToken, err := resolveSourceAccessToken(ctx, s.cfg.AccessTokenProvider, s.cfg.AccessToken, "personal source")
 	if err != nil {
+		// Transient provider failures (network, 429, 5xx) must not kill a
+		// long-running source; the reconnect loop retries after backoff.
+		if authpkg.ClassifyRefreshFailure(err) == authpkg.RefreshFailureTransient {
+			return nil, retryPersonal(err)
+		}
 		return nil, err
 	}
+	ticket, status, err := s.fetchTicketAttempt(ctx, accessToken)
+	if status == http.StatusUnauthorized && s.cfg.ForceRefreshToken != nil {
+		refreshed, refreshErr := refreshRejectedSourceToken(ctx, s.cfg.ForceRefreshToken, accessToken, "personal source", err)
+		if refreshErr != nil {
+			if authpkg.ClassifyRefreshFailure(refreshErr) == authpkg.RefreshFailureTransient {
+				return nil, retryPersonal(refreshErr)
+			}
+			return nil, refreshErr
+		}
+		// Retry once with the freshly rotated token; a second 401 stays fatal.
+		ticket, _, err = s.fetchTicketAttempt(ctx, refreshed)
+	}
+	return ticket, err
+}
+
+func (s *PersonalSource) fetchTicketAttempt(ctx context.Context, accessToken string) (*ticketResponse, int, error) {
 	body := map[string]any{
 		"sourceId": s.cfg.SourceID,
 		"mode":     s.cfg.TicketMode,
@@ -210,7 +240,7 @@ func (s *PersonalSource) fetchTicket(ctx context.Context) (*ticketResponse, erro
 	b, _ := json.Marshal(body)
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, s.cfg.TicketURL, bytes.NewReader(b))
 	if err != nil {
-		return nil, fmt.Errorf("personal source: create ticket request: %w", err)
+		return nil, 0, fmt.Errorf("personal source: create ticket request: %w", err)
 	}
 	req.Header.Set("Content-Type", "application/json")
 	req.Header.Set("Accept", "application/json")
@@ -221,28 +251,50 @@ func (s *PersonalSource) fetchTicket(ctx context.Context) (*ticketResponse, erro
 
 	resp, err := s.cfg.HTTPClient.Do(req)
 	if err != nil {
-		return nil, retryPersonal(fmt.Errorf("personal source: fetch ticket: %w", err))
+		return nil, 0, retryPersonal(fmt.Errorf("personal source: fetch ticket: %w", err))
 	}
 	defer resp.Body.Close()
-	data, err := io.ReadAll(io.LimitReader(resp.Body, config.MaxResponseBodySize))
-	if err != nil {
-		return nil, retryPersonal(fmt.Errorf("personal source: read ticket response: %w", err))
-	}
 	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		// Classify by status before touching the body: a truncated error body
+		// must not upgrade a fatal status (notably 401) into a retryable
+		// error, or the outer reconnect loop would bypass the single
+		// refresh-retry guard. The body is not used here, so drain it only
+		// best-effort for connection reuse.
+		_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, config.MaxResponseBodySize))
 		err := fmt.Errorf("personal source: ticket HTTP %d", resp.StatusCode)
 		if retryableTicketStatus(resp.StatusCode) {
-			return nil, retryPersonal(err)
+			return nil, resp.StatusCode, retryPersonal(err)
 		}
-		return nil, err
+		return nil, resp.StatusCode, err
+	}
+	data, err := io.ReadAll(io.LimitReader(resp.Body, config.MaxResponseBodySize))
+	if err != nil {
+		return nil, resp.StatusCode, retryPersonal(fmt.Errorf("personal source: read ticket response: %w", err))
 	}
 	ticket, err := decodeTicket(data)
 	if err != nil {
-		return nil, err
+		return nil, resp.StatusCode, err
 	}
 	if ticket.Endpoint == "" || ticket.Ticket == "" {
-		return nil, errors.New("personal source: ticket response missing endpoint or ticket")
+		return nil, resp.StatusCode, errors.New("personal source: ticket response missing endpoint or ticket")
 	}
-	return ticket, nil
+	return ticket, resp.StatusCode, nil
+}
+
+// refreshRejectedSourceToken funnels a server-side 401 into the optional
+// force-refresh callback. It hands the actual rejected token to the caller's
+// compare-and-refresh logic and returns the rotated token for an immediate
+// retry. Refresh failures keep the original 401 as context instead of being
+// dropped.
+func refreshRejectedSourceToken(ctx context.Context, refresh ForceRefreshTokenFn, rejectedToken, component string, cause error) (string, error) {
+	token, err := refresh(ctx, rejectedToken)
+	if err != nil {
+		return "", fmt.Errorf("%s: refresh rejected access token: %w", component, errors.Join(cause, err))
+	}
+	if token = strings.TrimSpace(token); token == "" {
+		return "", fmt.Errorf("%s: refresh rejected access token returned empty token: %w", component, cause)
+	}
+	return token, nil
 }
 
 func resolveSourceAccessToken(ctx context.Context, provider AccessTokenProvider, fallback, component string) (string, error) {
@@ -381,6 +433,13 @@ func isRetryablePersonalError(err error) bool {
 func personalRetryLogError(err error) string {
 	message := err.Error()
 	switch {
+	case strings.Contains(message, "resolve access token"), strings.Contains(message, "refresh rejected access token"):
+		// Token resolution/refresh errors may carry provider details; log
+		// only the structured HTTP status.
+		if status := refreshHTTPStatus(err); status != 0 {
+			return fmt.Sprintf("personal source: token refresh HTTP %d", status)
+		}
+		return "personal source: token refresh: temporary network error"
 	case strings.Contains(message, "ticket HTTP"):
 		return message
 	case strings.Contains(message, "fetch ticket"):
@@ -398,6 +457,14 @@ func personalRetryLogError(err error) string {
 	}
 }
 
+func refreshHTTPStatus(err error) int {
+	var statusErr *authpkg.HTTPStatusError
+	if !errors.As(err, &statusErr) || statusErr == nil {
+		return 0
+	}
+	return statusErr.StatusCode
+}
+
 func retryableTicketStatus(status int) bool {
 	return status == http.StatusRequestTimeout ||
 		status == http.StatusTooManyRequests ||

internal/event/source/portal_ticket.go

@@ -20,11 +20,13 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"log/slog"
 	"net/http"
 	"net/url"
 	"strings"
 	"time"
 
+	authpkg "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/auth"
 	dwsevent "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event"
 	"github.com/gorilla/websocket"
 	"github.com/open-dingtalk/dingtalk-stream-sdk-go/payload"
@@ -33,6 +35,8 @@ import (
 const (
 	PortalTicketModeNormal = "normal"
 	PortalTicketModeCustom = "custom"
+	portalReconnectMin     = time.Second
+	portalReconnectMax     = 30 * time.Second
 )
 
 // PortalTicketConfig describes the portal-managed user Stream ticket flow.
@@ -42,12 +46,45 @@ type PortalTicketConfig struct {
 	TicketURL           string
 	AccessToken         string
 	AccessTokenProvider AccessTokenProvider
+	ForceRefreshToken   ForceRefreshTokenFn
 	SourceID            string
 	Mode                string
 	ClientID            string
 	ClientSecret        string
 	UserAgent           string
 	HTTPClient          *http.Client
+	WebSocketDialer     *websocket.Dialer
+	ReconnectMin        time.Duration
+	ReconnectMax        time.Duration
+	DisableReconnect    bool
+}
+
+// portalStageError tags a portal stream failure with the stage it happened
+// in and whether the reconnect loop may retry it. Error() stays free of
+// untrusted response content so it is safe to log on every reconnect.
+type portalStageError struct {
+	stage     string
+	status    int
+	retryable bool
+	cause     error
+}
+
+func (e *portalStageError) Error() string {
+	if e == nil {
+		return "source: portal stream failed"
+	}
+	message := "source: portal " + strings.ReplaceAll(strings.TrimSpace(e.stage), "_", " ") + " failed"
+	if e.status != 0 {
+		message += fmt.Sprintf(" (HTTP %d)", e.status)
+	}
+	return message
+}
+
+func (e *portalStageError) Unwrap() error {
+	if e == nil {
+		return nil
+	}
+	return e.cause
 }
 
 var portalWriteMessage = func(conn *websocket.Conn, messageType int, data []byte) error {
@@ -97,48 +134,102 @@ func normalizePortalTicketMode(mode string) string {
 
 func (s *DingtalkSource) startPortalTicket(ctx context.Context, emit dwsevent.EmitFn) error {
 	s.machine.OnConnecting()
+	defer s.machine.OnStopped()
 
+	minBackoff := s.cfg.PortalTicket.ReconnectMin
+	if minBackoff <= 0 {
+		minBackoff = portalReconnectMin
+	}
+	maxBackoff := s.cfg.PortalTicket.ReconnectMax
+	if maxBackoff <= 0 {
+		maxBackoff = portalReconnectMax
+	}
+	if maxBackoff < minBackoff {
+		maxBackoff = minBackoff
+	}
+	backoff := minBackoff
+	for {
+		acked, err := s.runPortalTicketAttempt(ctx, emit)
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+		var stageErr *portalStageError
+		if !errors.As(err, &stageErr) || stageErr == nil || !stageErr.retryable || s.cfg.PortalTicket.DisableReconnect {
+			return err
+		}
+		if acked {
+			backoff = minBackoff
+		}
+		s.machine.OnReconnect()
+		slog.Warn("portal source reconnecting",
+			"stage", stageErr.stage,
+			"http_status", stageErr.status,
+			"error_type", fmt.Sprintf("%T", stageErr.cause),
+			"retry_in", backoff,
+			"reconnect_count", s.machine.Snapshot().ReconnectCount,
+		)
+		if err := waitPersonalReconnect(ctx, backoff); err != nil {
+			return err
+		}
+		backoff = nextPersonalBackoff(backoff, maxBackoff)
+	}
+}
+
+func (s *DingtalkSource) runPortalTicketAttempt(ctx context.Context, emit dwsevent.EmitFn) (bool, error) {
 	ticket, err := requestPortalTicket(ctx, s.cfg.PortalTicket)
 	if err != nil {
-		s.machine.OnStopped()
-		return err
+		return false, err
 	}
 	wsURL, err := websocketURL(ticket)
 	if err != nil {
-		s.machine.OnStopped()
-		return err
+		return false, err
 	}
 
 	userAgent := strings.TrimSpace(s.cfg.PortalTicket.UserAgent)
 	if userAgent == "" {
 		userAgent = "dws-event-consume"
 	}
-	conn, resp, err := (&websocket.Dialer{HandshakeTimeout: 20 * time.Second}).DialContext(ctx, wsURL, http.Header{
+	dialer := s.cfg.PortalTicket.WebSocketDialer
+	if dialer == nil {
+		dialer = &websocket.Dialer{HandshakeTimeout: 20 * time.Second}
+	}
+	conn, resp, err := dialer.DialContext(ctx, wsURL, http.Header{
 		"User-Agent": []string{userAgent},
 	})
 	if err != nil {
-		s.machine.OnStopped()
+		status := 0
+		cause := fmt.Errorf("source: portal stream connect: %w", err)
 		if resp != nil {
 			defer resp.Body.Close()
+			status = resp.StatusCode
 			raw, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
-			return fmt.Errorf("source: portal stream connect HTTP %d: %s: %w",
+			cause = fmt.Errorf("source: portal stream connect HTTP %d: %s: %w",
 				resp.StatusCode, truncatePortalTicketLog(string(raw), 300), err)
 		}
-		return fmt.Errorf("source: portal stream connect: %w", err)
+		return false, &portalStageError{
+			stage:     "stream_connect",
+			status:    status,
+			retryable: status == 0 || retryableTicketStatus(status),
+			cause:     cause,
+		}
 	}
-	defer conn.Close()
+	attemptCtx, cancel := context.WithCancel(ctx)
+	defer func() {
+		cancel()
+		_ = conn.Close()
+	}()
+	closeOnContext(attemptCtx, conn)
 	s.machine.OnConnected()
 
-	closeOnContext(ctx, conn)
 	handler := s.makeHandler(emit)
+	acked := false
 	for {
 		_, message, err := conn.ReadMessage()
 		if err != nil {
-			s.machine.OnStopped()
 			if isContextDone(ctx) {
-				return ctx.Err()
+				return acked, ctx.Err()
 			}
-			return fmt.Errorf("source: portal stream read: %w", err)
+			return acked, &portalStageError{stage: "stream_read", retryable: true, cause: fmt.Errorf("source: portal stream read: %w", err)}
 		}
 		df, err := payload.DecodeDataFrame(message)
 		if err != nil {
@@ -147,12 +238,12 @@ func (s *DingtalkSource) startPortalTicket(ctx context.Context, emit dwsevent.Em
 		resp, _ := handler(ctx, df)
 		ensurePortalAckHeaders(resp, df)
 		if err := portalWriteMessage(conn, websocket.TextMessage, resp.Encode()); err != nil {
-			s.machine.OnStopped()
 			if isContextDone(ctx) {
-				return ctx.Err()
+				return acked, ctx.Err()
 			}
-			return fmt.Errorf("source: portal stream ack: %w", err)
+			return acked, &portalStageError{stage: "stream_ack", retryable: true, cause: fmt.Errorf("source: portal stream ack: %w", err)}
 		}
+		acked = true
 	}
 }
 
@@ -164,12 +255,33 @@ type portalStreamTicket struct {
 func requestPortalTicket(ctx context.Context, cfg *PortalTicketConfig) (portalStreamTicket, error) {
 	accessToken, err := resolveSourceAccessToken(ctx, cfg.AccessTokenProvider, cfg.AccessToken, "source: portal ticket")
 	if err != nil {
+		// Transient provider failures (network, 429, 5xx) must not kill a
+		// long-running source; the reconnect loop retries after backoff.
+		if authpkg.ClassifyRefreshFailure(err) == authpkg.RefreshFailureTransient {
+			return portalStreamTicket{}, &portalStageError{stage: "ticket_auth", status: refreshHTTPStatus(err), retryable: true, cause: err}
+		}
 		return portalStreamTicket{}, err
 	}
 	httpClient := cfg.HTTPClient
 	if httpClient == nil {
 		httpClient = &http.Client{Timeout: 20 * time.Second}
 	}
+	ticket, status, err := requestPortalTicketAttempt(ctx, cfg, httpClient, accessToken)
+	if status == http.StatusUnauthorized && cfg.ForceRefreshToken != nil {
+		refreshed, refreshErr := refreshRejectedSourceToken(ctx, cfg.ForceRefreshToken, accessToken, "source: portal ticket", err)
+		if refreshErr != nil {
+			if authpkg.ClassifyRefreshFailure(refreshErr) == authpkg.RefreshFailureTransient {
+				return portalStreamTicket{}, &portalStageError{stage: "ticket_auth_refresh", status: refreshHTTPStatus(refreshErr), retryable: true, cause: refreshErr}
+			}
+			return portalStreamTicket{}, refreshErr
+		}
+		// Retry once with the freshly rotated token; a second 401 stays fatal.
+		ticket, _, err = requestPortalTicketAttempt(ctx, cfg, httpClient, refreshed)
+	}
+	return ticket, err
+}
+
+func requestPortalTicketAttempt(ctx context.Context, cfg *PortalTicketConfig, httpClient *http.Client, accessToken string) (portalStreamTicket, int, error) {
 	body := map[string]string{
 		"sourceId":    strings.TrimSpace(cfg.SourceID),
 		"channelType": strings.TrimSpace(cfg.SourceID),
@@ -182,7 +294,7 @@ func requestPortalTicket(ctx context.Context, cfg *PortalTicketConfig) (portalSt
 	rawBody, _ := json.Marshal(body)
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimSpace(cfg.TicketURL), bytes.NewReader(rawBody))
 	if err != nil {
-		return portalStreamTicket{}, err
+		return portalStreamTicket{}, 0, err
 	}
 	req.Header.Set("Content-Type", "application/json")
 	req.Header.Set("Accept", "application/json")
@@ -193,18 +305,34 @@ func requestPortalTicket(ctx context.Context, cfg *PortalTicketConfig) (portalSt
 
 	resp, err := httpClient.Do(req)
 	if err != nil {
-		return portalStreamTicket{}, fmt.Errorf("source: portal ticket request: %w", err)
+		return portalStreamTicket{}, 0, &portalStageError{stage: "ticket_request", retryable: true, cause: fmt.Errorf("source: portal ticket request: %w", err)}
 	}
 	defer resp.Body.Close()
-	raw, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
-	if resp.StatusCode >= 400 {
-		return portalStreamTicket{}, fmt.Errorf("source: portal ticket HTTP %d: %s",
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		// Preserve HTTP status semantics before touching the body. A truncated
+		// 401 body must not become a retryable read error that bypasses the
+		// single token-refresh guard; the body is only best-effort diagnostics.
+		raw, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
+		httpErr := fmt.Errorf("source: portal ticket HTTP %d: %s",
 			resp.StatusCode, truncatePortalTicketLog(string(raw), 300))
+		if retryableTicketStatus(resp.StatusCode) {
+			return portalStreamTicket{}, resp.StatusCode, &portalStageError{stage: "ticket_request", status: resp.StatusCode, retryable: true, cause: httpErr}
+		}
+		return portalStreamTicket{}, resp.StatusCode, httpErr
+	}
+	raw, readErr := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
+	if readErr != nil {
+		return portalStreamTicket{}, resp.StatusCode, &portalStageError{
+			stage:     "ticket_request",
+			status:    resp.StatusCode,
+			retryable: true,
+			cause:     fmt.Errorf("source: portal ticket read: %w", readErr),
+		}
 	}
 
 	var direct portalStreamTicket
 	if err := json.Unmarshal(raw, &direct); err == nil && direct.Endpoint != "" && direct.Ticket != "" {
-		return direct, nil
+		return direct, resp.StatusCode, nil
 	}
 
 	var envelope struct {
@@ -214,16 +342,16 @@ func requestPortalTicket(ctx context.Context, cfg *PortalTicketConfig) (portalSt
 		ErrorMsg  string             `json:"errorMsg"`
 	}
 	if err := json.Unmarshal(raw, &envelope); err != nil {
-		return portalStreamTicket{}, fmt.Errorf("source: portal ticket parse: %w", err)
+		return portalStreamTicket{}, resp.StatusCode, fmt.Errorf("source: portal ticket parse: %w", err)
 	}
 	if !envelope.Success {
-		return portalStreamTicket{}, fmt.Errorf("source: portal ticket failed: %s %s",
+		return portalStreamTicket{}, resp.StatusCode, fmt.Errorf("source: portal ticket failed: %s %s",
 			envelope.ErrorCode, envelope.ErrorMsg)
 	}
 	if envelope.Result.Endpoint == "" || envelope.Result.Ticket == "" {
-		return portalStreamTicket{}, errors.New("source: portal ticket result missing endpoint/ticket")
+		return portalStreamTicket{}, resp.StatusCode, errors.New("source: portal ticket result missing endpoint/ticket")
 	}
-	return envelope.Result, nil
+	return envelope.Result, resp.StatusCode, nil
 }
 
 func websocketURL(ticket portalStreamTicket) (string, error) {

internal/event/source/portal_ticket_refresh_test.go

@@ -0,0 +1,485 @@
+package source
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	dwsevent "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event"
+	"github.com/gorilla/websocket"
+	"github.com/open-dingtalk/dingtalk-stream-sdk-go/payload"
+)
+
+// TestCrossPlatformCoveragePortalStart401RefreshRetryEndToEnd drives the full production chain
+// DingtalkSource.Start → startPortalTicket → requestPortalTicket: the first
+// ticket request is rejected with 401, ForceRefreshToken rotates the token,
+// the in-chain retry succeeds with the fresh token and a WebSocket event is
+// delivered to emit.
+func TestCrossPlatformCoveragePortalStart401RefreshRetryEndToEnd(t *testing.T) {
+	var ticketCalls, refreshCalls atomic.Int64
+	var rejectedSeen atomic.Value
+
+	upgrader := websocket.Upgrader{}
+	var wsURL string
+	mux := http.NewServeMux()
+	mux.HandleFunc("/ticket", func(w http.ResponseWriter, r *http.Request) {
+		ticketCalls.Add(1)
+		switch r.Header.Get("x-user-access-token") {
+		case "fresh-token":
+			_ = json.NewEncoder(w).Encode(map[string]any{
+				"success": true,
+				"result":  map[string]string{"endpoint": wsURL, "ticket": "ticket-1"},
+			})
+		default:
+			w.WriteHeader(http.StatusUnauthorized)
+			_, _ = io.WriteString(w, "token expired")
+		}
+	})
+	mux.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) {
+		conn, err := upgrader.Upgrade(w, r, nil)
+		if err != nil {
+			return
+		}
+		defer conn.Close()
+		df := payload.DataFrame{Type: "event", Headers: payload.DataFrameHeader{payload.DataFrameHeaderKMessageId: "msg-1"}, Data: `{}`}
+		_ = conn.WriteJSON(df)
+		_, _, _ = conn.ReadMessage()
+	})
+	srv := httptest.NewServer(mux)
+	defer srv.Close()
+	wsURL = "ws" + strings.TrimPrefix(srv.URL, "http") + "/ws"
+
+	s, err := New(Config{PortalTicket: &PortalTicketConfig{
+		TicketURL: srv.URL + "/ticket",
+		AccessTokenProvider: func(context.Context) (string, error) {
+			return "stale-token", nil
+		},
+		ForceRefreshToken: func(_ context.Context, rejectedToken string) (string, error) {
+			refreshCalls.Add(1)
+			rejectedSeen.Store(rejectedToken)
+			return "fresh-token", nil
+		},
+		SourceID:   "source",
+		HTTPClient: srv.Client(),
+	}})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	emitted := make(chan struct{}, 1)
+	done := make(chan error, 1)
+	go func() { done <- s.Start(ctx, func(*dwsevent.RawEvent) { emitted <- struct{}{} }) }()
+	select {
+	case <-emitted:
+	case <-time.After(2 * time.Second):
+		t.Fatal("portal event timeout after 401 refresh retry")
+	}
+	cancel()
+	select {
+	case err := <-done:
+		if !errors.Is(err, context.Canceled) {
+			t.Fatalf("portal stop = %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("portal stop timeout")
+	}
+
+	if got := ticketCalls.Load(); got != 2 {
+		t.Fatalf("ticket calls = %d, want 2", got)
+	}
+	if got := refreshCalls.Load(); got != 1 {
+		t.Fatalf("refresh calls = %d, want 1", got)
+	}
+	if got, _ := rejectedSeen.Load().(string); got != "stale-token" {
+		t.Fatalf("rejected token = %q, want %q", got, "stale-token")
+	}
+}
+
+// TestCrossPlatformCoverageRequestPortalTicketRetryUsesRotatedTokenDirectly asserts the in-chain
+// retry sends the token returned by ForceRefreshToken instead of re-invoking
+// the provider (which could still serve the stale token).
+func TestCrossPlatformCoverageRequestPortalTicketRetryUsesRotatedTokenDirectly(t *testing.T) {
+	providerCalls := 0
+	var attemptTokens []string
+	client := &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
+		token := req.Header.Get("x-user-access-token")
+		attemptTokens = append(attemptTokens, token)
+		if token != "rotated" {
+			return &http.Response{StatusCode: http.StatusUnauthorized, Body: io.NopCloser(strings.NewReader("expired")), Header: make(http.Header)}, nil
+		}
+		return &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(`{"endpoint":"wss://x","ticket":"t"}`)), Header: make(http.Header)}, nil
+	})}
+	ticket, err := requestPortalTicket(context.Background(), &PortalTicketConfig{
+		TicketURL: "https://x",
+		AccessTokenProvider: func(context.Context) (string, error) {
+			providerCalls++
+			return "stale", nil
+		},
+		ForceRefreshToken: func(_ context.Context, rejectedToken string) (string, error) {
+			if rejectedToken != "stale" {
+				t.Fatalf("rejected token = %q, want %q", rejectedToken, "stale")
+			}
+			return "rotated", nil
+		},
+		SourceID:   "s",
+		HTTPClient: client,
+	})
+	if err != nil {
+		t.Fatalf("requestPortalTicket = %v", err)
+	}
+	if ticket.Endpoint != "wss://x" || ticket.Ticket != "t" {
+		t.Fatalf("ticket = %#v", ticket)
+	}
+	if providerCalls != 1 {
+		t.Fatalf("provider calls = %d, want 1", providerCalls)
+	}
+	if len(attemptTokens) != 2 || attemptTokens[0] != "stale" || attemptTokens[1] != "rotated" {
+		t.Fatalf("attempt tokens = %v", attemptTokens)
+	}
+}
+
+// TestCrossPlatformCoverageRequestPortalTicketRefreshFailureKeepsBothErrors asserts a failing
+// refresh neither retries nor drops the refresh error or the original 401.
+func TestCrossPlatformCoverageRequestPortalTicketRefreshFailureKeepsBothErrors(t *testing.T) {
+	refreshErr := errors.New("refresh_token exchange failed")
+	attempts := 0
+	client := &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
+		attempts++
+		return &http.Response{StatusCode: http.StatusUnauthorized, Body: io.NopCloser(strings.NewReader("expired")), Header: make(http.Header)}, nil
+	})}
+	_, err := requestPortalTicket(context.Background(), &PortalTicketConfig{
+		TicketURL:   "https://x",
+		AccessToken: "stale",
+		ForceRefreshToken: func(context.Context, string) (string, error) {
+			return "", refreshErr
+		},
+		SourceID:   "s",
+		HTTPClient: client,
+	})
+	if !errors.Is(err, refreshErr) {
+		t.Fatalf("error should wrap refresh error, got %v", err)
+	}
+	if err == nil || !strings.Contains(err.Error(), "HTTP 401") {
+		t.Fatalf("error should keep original 401, got %v", err)
+	}
+	if attempts != 1 {
+		t.Fatalf("attempts = %d, want 1 (no retry after failed refresh)", attempts)
+	}
+}
+
+// TestCrossPlatformCoverageRequestPortalTicketWithoutRefreshCallback401StaysFatal covers backward
+// compatibility: nil ForceRefreshToken keeps the single-attempt fatal 401.
+func TestCrossPlatformCoverageRequestPortalTicketWithoutRefreshCallback401StaysFatal(t *testing.T) {
+	attempts := 0
+	client := &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
+		attempts++
+		return &http.Response{StatusCode: http.StatusUnauthorized, Body: io.NopCloser(strings.NewReader("expired")), Header: make(http.Header)}, nil
+	})}
+	_, err := requestPortalTicket(context.Background(), &PortalTicketConfig{
+		TicketURL: "https://x", AccessToken: "stale", SourceID: "s", HTTPClient: client,
+	})
+	if err == nil || !strings.Contains(err.Error(), "HTTP 401") {
+		t.Fatalf("fatal 401 expected, got %v", err)
+	}
+	if attempts != 1 {
+		t.Fatalf("attempts = %d, want 1", attempts)
+	}
+}
+
+// TestCrossPlatformCoverageRequestPortalTicketSecond401IsFatal guards against refresh loops: the
+// controlled retry happens exactly once even if the rotated token is also
+// rejected.
+func TestCrossPlatformCoverageRequestPortalTicketSecond401IsFatal(t *testing.T) {
+	attempts := 0
+	refreshCalls := 0
+	client := &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
+		attempts++
+		return &http.Response{StatusCode: http.StatusUnauthorized, Body: io.NopCloser(strings.NewReader("expired")), Header: make(http.Header)}, nil
+	})}
+	_, err := requestPortalTicket(context.Background(), &PortalTicketConfig{
+		TicketURL:   "https://x",
+		AccessToken: "stale",
+		ForceRefreshToken: func(context.Context, string) (string, error) {
+			refreshCalls++
+			return "rotated-but-still-rejected", nil
+		},
+		SourceID:   "s",
+		HTTPClient: client,
+	})
+	if err == nil || !strings.Contains(err.Error(), "HTTP 401") {
+		t.Fatalf("fatal 401 expected after single retry, got %v", err)
+	}
+	if attempts != 2 || refreshCalls != 1 {
+		t.Fatalf("attempts = %d refreshCalls = %d, want 2/1", attempts, refreshCalls)
+	}
+}
+
+// TestCrossPlatformCoverageRequestPortalTicketRefreshEmptyTokenIsFatal asserts an empty rotated
+// token is rejected instead of being sent to the server.
+func TestCrossPlatformCoverageRequestPortalTicketRefreshEmptyTokenIsFatal(t *testing.T) {
+	attempts := 0
+	client := &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
+		attempts++
+		return &http.Response{StatusCode: http.StatusUnauthorized, Body: io.NopCloser(strings.NewReader("expired")), Header: make(http.Header)}, nil
+	})}
+	_, err := requestPortalTicket(context.Background(), &PortalTicketConfig{
+		TicketURL:   "https://x",
+		AccessToken: "stale",
+		ForceRefreshToken: func(context.Context, string) (string, error) {
+			return "  ", nil
+		},
+		SourceID:   "s",
+		HTTPClient: client,
+	})
+	if err == nil || !strings.Contains(err.Error(), "empty token") {
+		t.Fatalf("empty rotated token error expected, got %v", err)
+	}
+	if attempts != 1 {
+		t.Fatalf("attempts = %d, want 1", attempts)
+	}
+}
+
+// TestCrossPlatformCoveragePersonalFetchTicket401RefreshRetry mirrors the portal behavior for the
+// personal stream ticket path.
+func TestCrossPlatformCoveragePersonalFetchTicket401RefreshRetry(t *testing.T) {
+	var attemptTokens []string
+	src, err := NewPersonal(PersonalConfig{
+		AccessTokenProvider: func(context.Context) (string, error) { return "stale", nil },
+		ForceRefreshToken: func(_ context.Context, rejectedToken string) (string, error) {
+			if rejectedToken != "stale" {
+				t.Fatalf("rejected token = %q, want %q", rejectedToken, "stale")
+			}
+			return "rotated", nil
+		},
+		ClientID:  "client",
+		SourceID:  "source",
+		TicketURL: "https://ticket.test",
+		HTTPClient: &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
+			token := req.Header.Get("x-user-access-token")
+			attemptTokens = append(attemptTokens, token)
+			if token != "rotated" {
+				return &http.Response{StatusCode: http.StatusUnauthorized, Body: io.NopCloser(strings.NewReader("expired")), Header: make(http.Header)}, nil
+			}
+			return &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(`{"endpoint":"wss://stream.test","ticket":"ticket"}`)), Header: make(http.Header)}, nil
+		})},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	ticket, err := src.fetchTicket(context.Background())
+	if err != nil {
+		t.Fatalf("fetchTicket = %v", err)
+	}
+	if ticket.Endpoint != "wss://stream.test" || ticket.Ticket != "ticket" {
+		t.Fatalf("ticket = %#v", ticket)
+	}
+	if len(attemptTokens) != 2 || attemptTokens[0] != "stale" || attemptTokens[1] != "rotated" {
+		t.Fatalf("attempt tokens = %v", attemptTokens)
+	}
+}
+
+// TestCrossPlatformCoveragePersonalFetchTicket401RefreshFailureStaysFatal asserts a failed refresh
+// keeps the 401 fatal (not retryable) and wraps the refresh error.
+func TestCrossPlatformCoveragePersonalFetchTicket401RefreshFailureStaysFatal(t *testing.T) {
+	refreshErr := errors.New("refresh_token exchange failed")
+	src, err := NewPersonal(PersonalConfig{
+		AccessToken: "stale",
+		ForceRefreshToken: func(context.Context, string) (string, error) {
+			return "", refreshErr
+		},
+		ClientID:  "client",
+		SourceID:  "source",
+		TicketURL: "https://ticket.test",
+		HTTPClient: &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
+			return &http.Response{StatusCode: http.StatusUnauthorized, Body: io.NopCloser(strings.NewReader("expired")), Header: make(http.Header)}, nil
+		})},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = src.fetchTicket(context.Background())
+	if !errors.Is(err, refreshErr) {
+		t.Fatalf("error should wrap refresh error, got %v", err)
+	}
+	if isRetryablePersonalError(err) {
+		t.Fatalf("failed refresh should stay fatal, got retryable %v", err)
+	}
+}
+
+// brokenBody simulates a response body that fails mid-read, e.g. the server
+// closing the connection before the error payload is fully written.
+type brokenBody struct{}
+
+func (brokenBody) Read([]byte) (int, error) { return 0, io.ErrUnexpectedEOF }
+func (brokenBody) Close() error             { return nil }
+
+func TestCrossPlatformCoveragePortalTicketNon2xxTruncatedBodyKeepsStatus(t *testing.T) {
+	for _, tc := range []struct {
+		status    int
+		retryable bool
+	}{
+		{status: http.StatusUnauthorized},
+		{status: http.StatusTooManyRequests, retryable: true},
+		{status: http.StatusServiceUnavailable, retryable: true},
+	} {
+		t.Run(http.StatusText(tc.status), func(t *testing.T) {
+			client := &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
+				return &http.Response{StatusCode: tc.status, Body: brokenBody{}, Header: make(http.Header)}, nil
+			})}
+
+			_, status, err := requestPortalTicketAttempt(context.Background(), &PortalTicketConfig{
+				TicketURL: "https://ticket.test",
+				SourceID:  "source",
+			}, client, "token")
+			if status != tc.status || err == nil || !strings.Contains(err.Error(), fmt.Sprintf("HTTP %d", tc.status)) {
+				t.Fatalf("requestPortalTicketAttempt() status=%d err=%v, want HTTP %d", status, err, tc.status)
+			}
+			if errors.Is(err, io.ErrUnexpectedEOF) {
+				t.Fatalf("non-2xx status error should not expose diagnostic body read failure: %v", err)
+			}
+			var stageErr *portalStageError
+			if tc.retryable {
+				if !errors.As(err, &stageErr) || stageErr.status != tc.status || !stageErr.retryable {
+					t.Fatalf("retryable status should return retryable portalStageError, got %T %v", err, err)
+				}
+			} else if errors.As(err, &stageErr) && stageErr.retryable {
+				t.Fatalf("fatal status should not become retryable stage error: %v", err)
+			}
+		})
+	}
+}
+
+func TestCrossPlatformCoveragePortalTicket200TruncatedBodyIsRetryable(t *testing.T) {
+	client := &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
+		return &http.Response{StatusCode: http.StatusOK, Body: brokenBody{}, Header: make(http.Header)}, nil
+	})}
+
+	_, status, err := requestPortalTicketAttempt(context.Background(), &PortalTicketConfig{
+		TicketURL: "https://ticket.test",
+		SourceID:  "source",
+	}, client, "token")
+	if status != http.StatusOK || !errors.Is(err, io.ErrUnexpectedEOF) {
+		t.Fatalf("requestPortalTicketAttempt() status=%d err=%v, want 200 + io.ErrUnexpectedEOF", status, err)
+	}
+	var stageErr *portalStageError
+	if !errors.As(err, &stageErr) || !stageErr.retryable || stageErr.stage != "ticket_request" {
+		t.Fatalf("2xx truncated body should be retryable ticket_request stage, got %T %v", err, err)
+	}
+}
+
+// TestCrossPlatformCoveragePersonalFetchTicket401TruncatedBodyStaysFatal guards the single
+// refresh-retry protection: a 401 whose body fails with unexpected EOF must
+// be classified by status (fatal) and never wrapped as retryable, otherwise
+// the outer reconnect loop would refresh again on every iteration.
+func TestCrossPlatformCoveragePersonalFetchTicket401TruncatedBodyStaysFatal(t *testing.T) {
+	attempts := 0
+	refreshCalls := 0
+	src, err := NewPersonal(PersonalConfig{
+		AccessToken: "stale",
+		ForceRefreshToken: func(context.Context, string) (string, error) {
+			refreshCalls++
+			return "rotated", nil
+		},
+		ClientID:  "client",
+		SourceID:  "source",
+		TicketURL: "https://ticket.test",
+		HTTPClient: &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
+			attempts++
+			return &http.Response{StatusCode: http.StatusUnauthorized, Body: brokenBody{}, Header: make(http.Header)}, nil
+		})},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = src.fetchTicket(context.Background())
+	if err == nil || !strings.Contains(err.Error(), "HTTP 401") {
+		t.Fatalf("fatal 401 error expected, got %v", err)
+	}
+	if isRetryablePersonalError(err) {
+		t.Fatalf("401 with truncated body must stay fatal, got retryable %v", err)
+	}
+	if refreshCalls != 1 {
+		t.Fatalf("refresh calls = %d, want 1", refreshCalls)
+	}
+	if attempts != 2 {
+		t.Fatalf("attempts = %d, want 2", attempts)
+	}
+}
+
+// TestCrossPlatformCoveragePersonalFetchTicket200TruncatedBodyStaysRetryable pins the existing
+// behavior for success responses: a body read failure on 2xx is a transient
+// transport problem and remains retryable.
+func TestCrossPlatformCoveragePersonalFetchTicket200TruncatedBodyStaysRetryable(t *testing.T) {
+	src, err := NewPersonal(PersonalConfig{
+		AccessToken: "token",
+		ClientID:    "client",
+		SourceID:    "source",
+		TicketURL:   "https://ticket.test",
+		HTTPClient: &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
+			return &http.Response{StatusCode: 200, Body: brokenBody{}, Header: make(http.Header)}, nil
+		})},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = src.fetchTicket(context.Background())
+	if !errors.Is(err, io.ErrUnexpectedEOF) {
+		t.Fatalf("error should wrap io.ErrUnexpectedEOF, got %v", err)
+	}
+	if !isRetryablePersonalError(err) {
+		t.Fatalf("2xx body read failure should stay retryable, got %v", err)
+	}
+}
+
+// TestCrossPlatformCoveragePersonalFetchTicketAttemptTransportAndPayloadEdges covers the
+// remaining fetchTicketAttempt branches: transport failures and retryable
+// statuses stay retryable, while a well-formed response missing the endpoint
+// or ticket fields stays fatal.
+func TestCrossPlatformCoveragePersonalFetchTicketAttemptTransportAndPayloadEdges(t *testing.T) {
+	newSource := func(rt roundTripFunc) *PersonalSource {
+		src, err := NewPersonal(PersonalConfig{
+			AccessToken: "token",
+			ClientID:    "client",
+			SourceID:    "source",
+			TicketURL:   "https://ticket.test",
+			HTTPClient:  &http.Client{Transport: rt},
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		return src
+	}
+
+	dialErr := errors.New("dial tcp: connection refused")
+	src := newSource(func(*http.Request) (*http.Response, error) { return nil, dialErr })
+	_, _, err := src.fetchTicketAttempt(context.Background(), "token")
+	if !errors.Is(err, dialErr) || !isRetryablePersonalError(err) {
+		t.Fatalf("transport failure should stay retryable and wrap cause, got %v", err)
+	}
+
+	src = newSource(func(*http.Request) (*http.Response, error) {
+		return &http.Response{StatusCode: http.StatusServiceUnavailable, Body: io.NopCloser(strings.NewReader("busy")), Header: make(http.Header)}, nil
+	})
+	_, status, err := src.fetchTicketAttempt(context.Background(), "token")
+	if status != http.StatusServiceUnavailable || !isRetryablePersonalError(err) {
+		t.Fatalf("503 should stay retryable, got status %d err %v", status, err)
+	}
+
+	src = newSource(func(*http.Request) (*http.Response, error) {
+		return &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(`{"endpoint":"","ticket":""}`)), Header: make(http.Header)}, nil
+	})
+	_, _, err = src.fetchTicketAttempt(context.Background(), "token")
+	if err == nil || isRetryablePersonalError(err) || !strings.Contains(err.Error(), "missing endpoint or ticket") {
+		t.Fatalf("missing ticket fields should stay fatal, got %v", err)
+	}
+}

internal/event/source/transient_auth_reconnect_test.go

@@ -0,0 +1,295 @@
+// Copyright 2026 Alibaba Group
+// Licensed under the Apache License, Version 2.0 (the "License");
+
+package source
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	authpkg "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/auth"
+	dwsevent "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event"
+	"github.com/gorilla/websocket"
+	"github.com/open-dingtalk/dingtalk-stream-sdk-go/payload"
+)
+
+func transientRefreshError() error {
+	return &authpkg.HTTPStatusError{StatusCode: http.StatusServiceUnavailable}
+}
+
+func terminalRefreshError() error {
+	return &authpkg.HTTPStatusError{StatusCode: http.StatusUnauthorized}
+}
+
+func TestCrossPlatformCoveragePersonalRetryLogErrorReportsOnlySafeTransientStatus(t *testing.T) {
+	cause := errors.New("personal source: ticket HTTP 401 secret detail")
+	_, err := refreshRejectedSourceToken(context.Background(), func(context.Context, string) (string, error) {
+		return "", transientRefreshError()
+	}, "rejected", "personal source", cause)
+	if got, want := personalRetryLogError(retryPersonal(err)), "personal source: token refresh HTTP 503"; got != want {
+		t.Fatalf("personalRetryLogError() = %q, want %q", got, want)
+	}
+}
+
+func TestCrossPlatformCoveragePersonalSourceRetriesTransientTokenResolutionFailure(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	var calls atomic.Int32
+	src, err := NewPersonal(PersonalConfig{
+		AccessTokenProvider: func(context.Context) (string, error) {
+			if calls.Add(1) == 2 {
+				cancel()
+			}
+			return "", transientRefreshError()
+		},
+		ClientID:     "client",
+		SourceID:     "open",
+		TicketURL:    "https://ticket.invalid",
+		ReconnectMin: time.Millisecond,
+		ReconnectMax: time.Millisecond,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = src.Start(ctx, func(*dwsevent.RawEvent) {})
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("Start() error = %v, want context canceled after retry", err)
+	}
+	if calls.Load() != 2 || src.State().ReconnectCount != 1 {
+		t.Fatalf("provider calls=%d reconnects=%d, want 2 calls and 1 reconnect", calls.Load(), src.State().ReconnectCount)
+	}
+}
+
+func TestCrossPlatformCoveragePersonalSourceDoesNotRetryTerminalTokenResolutionFailure(t *testing.T) {
+	var calls atomic.Int32
+	src, err := NewPersonal(PersonalConfig{
+		AccessTokenProvider: func(context.Context) (string, error) {
+			calls.Add(1)
+			return "", terminalRefreshError()
+		},
+		ClientID:  "client",
+		SourceID:  "open",
+		TicketURL: "https://ticket.invalid",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = src.Start(context.Background(), func(*dwsevent.RawEvent) {})
+	if authpkg.ClassifyRefreshFailure(err) != authpkg.RefreshFailureTerminal {
+		t.Fatalf("Start() error = %v, want terminal refresh failure", err)
+	}
+	if calls.Load() != 1 || src.State().ReconnectCount != 0 {
+		t.Fatalf("provider calls=%d reconnects=%d, want 1 call and no reconnect", calls.Load(), src.State().ReconnectCount)
+	}
+}
+
+func TestCrossPlatformCoveragePersonalSourceRetriesTransientRejectedTokenRefresh(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	var ticketCalls atomic.Int32
+	var refreshCalls atomic.Int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		ticketCalls.Add(1)
+		w.WriteHeader(http.StatusUnauthorized)
+	}))
+	defer srv.Close()
+
+	src, err := NewPersonal(PersonalConfig{
+		AccessTokenProvider: func(context.Context) (string, error) { return "old-token", nil },
+		ForceRefreshToken: func(_ context.Context, rejected string) (string, error) {
+			if rejected != "old-token" {
+				t.Fatalf("rejected token = %q, want old-token", rejected)
+			}
+			if refreshCalls.Add(1) == 2 {
+				cancel()
+			}
+			return "", transientRefreshError()
+		},
+		ClientID:     "client",
+		SourceID:     "open",
+		TicketURL:    srv.URL,
+		HTTPClient:   srv.Client(),
+		ReconnectMin: time.Millisecond,
+		ReconnectMax: time.Millisecond,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = src.Start(ctx, func(*dwsevent.RawEvent) {})
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("Start() error = %v, want context canceled after retry", err)
+	}
+	if ticketCalls.Load() != 2 || refreshCalls.Load() != 2 || src.State().ReconnectCount != 1 {
+		t.Fatalf("ticket calls=%d refresh calls=%d reconnects=%d", ticketCalls.Load(), refreshCalls.Load(), src.State().ReconnectCount)
+	}
+}
+
+func TestCrossPlatformCoveragePersonalRetryLogErrorFallsBackToNetworkMessage(t *testing.T) {
+	err := retryPersonal(fmt.Errorf("personal source: resolve access token: %w", errors.New("dial tcp: lookup oauth.invalid")))
+	if got, want := personalRetryLogError(err), "personal source: token refresh: temporary network error"; got != want {
+		t.Fatalf("personalRetryLogError() = %q, want %q", got, want)
+	}
+}
+
+func TestCrossPlatformCoveragePortalStageErrorNilAndUnwrap(t *testing.T) {
+	var nilErr *portalStageError
+	if got, want := nilErr.Error(), "source: portal stream failed"; got != want {
+		t.Fatalf("nil stage error = %q, want %q", got, want)
+	}
+	if nilErr.Unwrap() != nil {
+		t.Fatal("nil stage error should unwrap to nil")
+	}
+	cause := errors.New("cause")
+	stageErr := &portalStageError{stage: "stream_read", retryable: true, cause: cause}
+	if !errors.Is(stageErr, cause) {
+		t.Fatalf("stage error should unwrap to cause: %v", stageErr)
+	}
+}
+
+func TestCrossPlatformCoveragePortalSourceRetriesTransientTokenResolutionFailure(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	var calls atomic.Int32
+	src, err := New(Config{PortalTicket: &PortalTicketConfig{
+		TicketURL: "https://ticket.invalid",
+		AccessTokenProvider: func(context.Context) (string, error) {
+			if calls.Add(1) == 2 {
+				cancel()
+			}
+			return "", transientRefreshError()
+		},
+		SourceID: "open",
+		// Min above max exercises the reconnect clamp.
+		ReconnectMin: 2 * time.Millisecond,
+		ReconnectMax: time.Millisecond,
+	}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = src.Start(ctx, func(*dwsevent.RawEvent) {})
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("Start() error = %v, want context canceled after retry", err)
+	}
+	if calls.Load() != 2 || src.State().ReconnectCount != 1 {
+		t.Fatalf("provider calls=%d reconnects=%d, want 2 calls and 1 reconnect", calls.Load(), src.State().ReconnectCount)
+	}
+}
+
+func TestCrossPlatformCoveragePortalSourceResetsBackoffAfterAckedAttempt(t *testing.T) {
+	var ticketCalls atomic.Int32
+	upgrader := websocket.Upgrader{}
+	var wsURL string
+	mux := http.NewServeMux()
+	mux.HandleFunc("/ticket", func(w http.ResponseWriter, _ *http.Request) {
+		if ticketCalls.Add(1) > 1 {
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+		_, _ = io.WriteString(w, `{"endpoint":`+strconvQuote(wsURL)+`,"ticket":"t"}`)
+	})
+	mux.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) {
+		conn, err := upgrader.Upgrade(w, r, nil)
+		if err != nil {
+			return
+		}
+		defer conn.Close()
+		df := payload.DataFrame{Type: "event", Headers: payload.DataFrameHeader{payload.DataFrameHeaderKMessageId: "m"}, Data: `{}`}
+		_ = conn.WriteJSON(df)
+		// Wait for the ACK, then close so the read fails retryably with an
+		// acked attempt behind it, which resets the reconnect backoff.
+		_, _, _ = conn.ReadMessage()
+	})
+	srv := httptest.NewServer(mux)
+	defer srv.Close()
+	wsURL = "ws" + strings.TrimPrefix(srv.URL, "http") + "/ws"
+
+	src, err := New(Config{PortalTicket: &PortalTicketConfig{
+		TicketURL:    srv.URL + "/ticket",
+		AccessToken:  "t",
+		SourceID:     "open",
+		HTTPClient:   srv.Client(),
+		ReconnectMin: time.Millisecond,
+		ReconnectMax: time.Millisecond,
+	}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var events atomic.Int32
+	err = src.Start(context.Background(), func(*dwsevent.RawEvent) { events.Add(1) })
+	if err == nil || !strings.Contains(err.Error(), "portal ticket HTTP 400") {
+		t.Fatalf("Start() error = %v, want fatal ticket HTTP 400 after reconnect", err)
+	}
+	if events.Load() != 1 || ticketCalls.Load() != 2 || src.State().ReconnectCount != 1 {
+		t.Fatalf("events=%d ticket calls=%d reconnects=%d, want 1/2/1", events.Load(), ticketCalls.Load(), src.State().ReconnectCount)
+	}
+}
+
+func TestCrossPlatformCoveragePortalSourceDoesNotRetryTerminalTokenResolutionFailure(t *testing.T) {
+	var calls atomic.Int32
+	src, err := New(Config{PortalTicket: &PortalTicketConfig{
+		TicketURL: "https://ticket.invalid",
+		AccessTokenProvider: func(context.Context) (string, error) {
+			calls.Add(1)
+			return "", terminalRefreshError()
+		},
+		SourceID: "open",
+	}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = src.Start(context.Background(), func(*dwsevent.RawEvent) {})
+	if authpkg.ClassifyRefreshFailure(err) != authpkg.RefreshFailureTerminal {
+		t.Fatalf("Start() error = %v, want terminal refresh failure", err)
+	}
+	if calls.Load() != 1 || src.State().ReconnectCount != 0 {
+		t.Fatalf("provider calls=%d reconnects=%d, want 1 call and no reconnect", calls.Load(), src.State().ReconnectCount)
+	}
+}
+
+func TestCrossPlatformCoveragePortalSourceRetriesTransientRejectedTokenRefresh(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	var ticketCalls atomic.Int32
+	var refreshCalls atomic.Int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		ticketCalls.Add(1)
+		w.WriteHeader(http.StatusUnauthorized)
+	}))
+	defer srv.Close()
+
+	src, err := New(Config{PortalTicket: &PortalTicketConfig{
+		TicketURL:           srv.URL,
+		AccessTokenProvider: func(context.Context) (string, error) { return "old-token", nil },
+		ForceRefreshToken: func(_ context.Context, rejected string) (string, error) {
+			if rejected != "old-token" {
+				t.Fatalf("rejected token = %q, want old-token", rejected)
+			}
+			if refreshCalls.Add(1) == 2 {
+				cancel()
+			}
+			return "", transientRefreshError()
+		},
+		SourceID:     "open",
+		HTTPClient:   srv.Client(),
+		ReconnectMin: time.Millisecond,
+		ReconnectMax: time.Millisecond,
+	}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = src.Start(ctx, func(*dwsevent.RawEvent) {})
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("Start() error = %v, want context canceled after retry", err)
+	}
+	if ticketCalls.Load() != 2 || refreshCalls.Load() != 2 || src.State().ReconnectCount != 1 {
+		t.Fatalf("ticket calls=%d refresh calls=%d reconnects=%d", ticketCalls.Load(), refreshCalls.Load(), src.State().ReconnectCount)
+	}
+}