Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 19 additions & 1 deletion cmd/subrouter/sr_cloud.go
Original file line number Diff line number Diff line change
Expand Up @@ -771,17 +771,35 @@ func (r srRunner) cloudAccountImport(
)
}
uploaded := 0
migrated := 0
for _, upload := range pending {
if _, err := client.UploadAccount(ctx, upload.body); err != nil {
return fmt.Errorf("upload %s %s: %w", upload.kind, upload.label, err)
}
uploaded++
fmt.Fprintf(r.out, "uploaded %-20s %s\n", upload.kind, upload.label)
// Hand over ownership as well as the bytes. The provider rotates the
// refresh token on every use, so leaving a refreshable local copy makes
// this machine and the vault invalidate each other's chain. The record
// is kept for rollback, just where the daemon will not refresh it.
if upload.kind != "codex" {
continue
}
path, ok, err := r.store.MigrateStoredAway(upload.label)
if err != nil {
return fmt.Errorf("hand over %s: %w", upload.label, err)
Comment on lines 776 to +790

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift

Make remote adoption and local disowning failure-safe.

UploadAccount adopts the credential before MigrateStoredAway. If the move fails, the vault has accepted the refresh chain while the local active file remains, recreating the two-refreshers condition this PR prevents. Stage the upload until migration succeeds, or compensate by disabling/deleting the newly uploaded broker account before returning the error; cover this failure path with a test.

🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@cmd/subrouter/sr_cloud.go` around lines 776 - 790, Make the UploadAccount and
MigrateStoredAway sequence in the account upload flow failure-safe: ensure local
disowning succeeds before leaving the remote account active, or compensate by
disabling/deleting the newly uploaded broker account if migration fails.
Preserve rollback data and return the migration error, and add a test covering
migration failure to verify no active remote adoption remains while the local
credential is still present.

}
if ok {
migrated++
fmt.Fprintf(r.out, " local copy kept as a record at %s\n", path)
}
}
fmt.Fprintf(
r.out,
"\nUploaded %d shared account(s). Central refresh adoption may rotate OAuth chains; local files remain only as migration records and may require re-authentication for rollback.\n",
"\nUploaded %d shared account(s); %d local credential(s) handed over. This machine no longer refreshes them, so the vault owns their token chains. Rollback records remain in %s.\n",
uploaded,
migrated,
filepath.Join(r.store.StoreDir(), accounts.MigratedDirName),
)
return restartInstalledDaemon()
}
Expand Down
26 changes: 26 additions & 0 deletions internal/accounts/codex_store.go
Original file line number Diff line number Diff line change
Expand Up @@ -292,6 +292,32 @@ func (s CodexStore) RemoveStored(identifier string) (StoredCodexAccount, bool, e
return account, true, nil
}

// MigratedDirName holds credentials this machine has handed to the team vault.
// The store does not scan it, so the local daemon stops refreshing them: the
// provider rotates refresh tokens on use, so a credential must have exactly one
// refresher or the two invalidate each other.
const MigratedDirName = "migrated"

// MigrateStoredAway moves a stored account out of the active store and into the
// migrated directory, keeping it as a rollback record the daemon will not touch.
// It returns the path the record now lives at.
func (s CodexStore) MigrateStoredAway(identifier string) (string, bool, error) {
account, ok, err := s.FindStored(identifier)
if err != nil || !ok {
return "", ok, err
}
name := emailToFilename(account.Email)
dest := filepath.Join(s.Dir, MigratedDirName)
if err := os.MkdirAll(dest, 0o700); err != nil {
return "", false, err
}
target := filepath.Join(dest, name)
if err := os.Rename(filepath.Join(s.Dir, name), target); err != nil {
return "", false, err
}
return target, true, nil
}

func emailToFilename(email string) string {
var b strings.Builder
for _, r := range email {
Expand Down
66 changes: 66 additions & 0 deletions internal/accounts/codex_store_migrate_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,66 @@
package accounts

import (
"os"
"path/filepath"
"testing"
)

// A credential handed to the team vault must stop being refreshed here: the
// provider rotates refresh tokens on use, so two refreshers invalidate each
// other. Moving the file out of the scanned store is what disowns it, while
// keeping it readable for rollback.
func TestMigrateStoredAwayDisownsButKeepsRecord(t *testing.T) {
store := CodexStore{Dir: t.TempDir()}
account := StoredCodexAccount{
Email: "lc+8@cmux.com",
Auth: CodexAuthFile{Tokens: &CodexTokens{
AccessToken: "access",
RefreshToken: "refresh",
IDToken: "id",
}},
}
if err := store.SaveStored(account); err != nil {
t.Fatal(err)
}
if stored, err := store.ListStored(); err != nil || len(stored) != 1 {
t.Fatalf("precondition: stored=%d err=%v", len(stored), err)
}

path, ok, err := store.MigrateStoredAway("lc+8@cmux.com")
if err != nil || !ok {
t.Fatalf("migrate: ok=%v err=%v", ok, err)
}

stored, err := store.ListStored()
if err != nil {
t.Fatal(err)
}
if len(stored) != 0 {
t.Fatalf("account still in the active store after handover: %+v", stored)
}
if filepath.Dir(path) != filepath.Join(store.Dir, MigratedDirName) {
t.Fatalf("record landed at %s, want it under %s", path, MigratedDirName)
}
body, err := os.ReadFile(path)
if err != nil {
t.Fatalf("rollback record unreadable: %v", err)
}
if len(body) == 0 {
t.Fatal("rollback record is empty")
}
info, err := os.Stat(filepath.Dir(path))
if err != nil {
t.Fatal(err)
}
if perm := info.Mode().Perm(); perm != 0o700 {
t.Fatalf("record directory mode %o, want 700", perm)
}
}

func TestMigrateStoredAwayReportsMissingAccount(t *testing.T) {
store := CodexStore{Dir: t.TempDir()}
if _, ok, err := store.MigrateStoredAway("nobody@example.com"); ok || err != nil {
t.Fatalf("ok=%v err=%v, want false and no error", ok, err)
}
}
2 changes: 1 addition & 1 deletion internal/proxy/dashboard_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -8,8 +8,8 @@ import (
"strings"
"testing"

"github.com/manaflow-ai/subrouter/session"
"github.com/manaflow-ai/subrouter/internal/transcript"
"github.com/manaflow-ai/subrouter/session"
)

func TestDashboardAndTranscriptEndpoints(t *testing.T) {
Expand Down
13 changes: 13 additions & 0 deletions internal/proxy/proxy.go
Original file line number Diff line number Diff line change
Expand Up @@ -1195,6 +1195,14 @@ func (s Server) reloadAccounts(ctx context.Context) (int, int, error) {
if s.SchedulerRef == nil {
return len(loaded), 0, nil
}
// In team mode the vault owns these credentials and the broker picks the
// account, so local scoring is both unnecessary and destructive: scoring
// refreshes every OAuth account, the provider rotates the refresh token on
// every use, and a second refresher invalidates the vault's chain (and vice
// versa). Two refreshers permanently break each other's accounts.
if s.CredentialBroker != nil {
return len(loaded), 0, nil
}
scores, scored := s.scoreAccounts(ctx, loaded)
if scored == 0 {
if s.Logger != nil {
Expand Down Expand Up @@ -3364,6 +3372,11 @@ func codexResponsePath(path string) bool {
// wholesale: a codex-triggered refresh must not wipe claude scores or vice
// versa.
func (s Server) refreshUsageScoresIfStale(ctx context.Context) {
// See reloadAccounts: in team mode refreshing local OAuth accounts rotates
// refresh tokens the vault owns, which invalidates them for both sides.
if s.CredentialBroker != nil {
return
}
if s.SchedulerRef == nil || !s.SchedulerRef.BeginRefreshIfStale(s.UsageScoreTTL) {
return
}
Expand Down
93 changes: 93 additions & 0 deletions internal/proxy/team_mode_refresh_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,93 @@
package proxy

import (
"context"
"sync/atomic"
"testing"

"github.com/manaflow-ai/subrouter/internal/accounts"
"github.com/manaflow-ai/subrouter/selectacct"
)

// In team mode the vault owns the credentials and refreshes them. The provider
// rotates the refresh token on every use, so a second refresher invalidates the
// vault's chain and the vault invalidates this machine's: both sides end up
// dead. Observed in production across a laptop and a mac mini sharing one
// account store, where 11 of 12 accounts failed with refresh_token_reused.
//
// So a server with a credential broker must never refresh a local OAuth
// account, on any path.
func TestTeamModeNeverRefreshesLocalAccounts(t *testing.T) {
local := []accounts.Account{
{ID: "a@example.com", Provider: accounts.ProviderCodex, AuthMode: accounts.AuthModeOAuth, Token: "a"},
{ID: "b@example.com", Provider: accounts.ProviderCodex, AuthMode: accounts.AuthModeOAuth, Token: "b"},
}

// The store must contain the accounts, not just the in-memory slice:
// reloadAccounts re-reads it, so a test that relies on whatever the developer
// happens to have in ~/.subrouter passes locally and fails in CI.
newStore := func() accounts.CodexStore {
store := accounts.CodexStore{Dir: t.TempDir()}
for _, account := range local {
if err := store.SaveStored(accounts.StoredCodexAccount{
Email: account.ID,
Provider: accounts.ProviderCodex,
Auth: accounts.CodexAuthFile{Tokens: &accounts.CodexTokens{
AccessToken: "access-" + account.ID,
RefreshToken: "refresh-" + account.ID,
IDToken: "id-" + account.ID,
}},
}); err != nil {
t.Fatalf("seed store: %v", err)
}
}
return store
}

newServer := func(refreshes *int32, credBroker CredentialBroker) Server {
return Server{
Accounts: local,
SchedulerRef: selectacct.NewSchedulerRef(selectacct.NewScheduler(nil)),
AccountRef: NewAccountRef(newStore(), local, nil),
RefreshAccountFn: func(_ context.Context, a accounts.Account) (accounts.Account, error) {
atomic.AddInt32(refreshes, 1)
return a, nil
},
CredentialBroker: credBroker,
UsageScoreTTL: 0,
}
}

t.Run("reloadAccounts", func(t *testing.T) {
var teamRefreshes int32
team := newServer(&teamRefreshes, &fakeCredentialBroker{})
if _, _, err := team.reloadAccounts(context.Background()); err != nil {
t.Fatalf("reload in team mode: %v", err)
}
if got := atomic.LoadInt32(&teamRefreshes); got != 0 {
t.Fatalf("team mode refreshed %d local accounts, want 0", got)
}
})

t.Run("refreshUsageScoresIfStale", func(t *testing.T) {
var teamRefreshes int32
team := newServer(&teamRefreshes, &fakeCredentialBroker{})
team.refreshUsageScoresIfStale(context.Background())
if got := atomic.LoadInt32(&teamRefreshes); got != 0 {
t.Fatalf("team mode refreshed %d local accounts while scoring, want 0", got)
}
})

// Local mode is the other half of the contract: with no broker this machine
// is the only refresher, so it must still refresh or scores go stale.
t.Run("local mode still refreshes", func(t *testing.T) {
var localRefreshes int32
solo := newServer(&localRefreshes, nil)
if _, _, err := solo.reloadAccounts(context.Background()); err != nil {
t.Fatalf("reload in local mode: %v", err)
}
if got := atomic.LoadInt32(&localRefreshes); got == 0 {
t.Fatal("local mode refreshed nothing; the gate is too wide")
}
})
}