From f78ef234210b384ea426b426746dc499a3c3610c Mon Sep 17 00:00:00 2001 From: Rob Zolkos Date: Tue, 13 Oct 2026 17:56:53 -0400 Subject: [PATCH 1/5] One connector per profile: key the state folder and lock by profile The connector's state folder was -, so its instance lock and ledger belonged to the agent: two profiles for the same agent on one computer could not each run a connector. The folder is now -- (and the same under connect-shadow), so the lock and the ledger that holds the feed cursor are per profile. A second connector on the same profile is still refused. The old - folder carries over by rename to the first profile that starts and proves it is that agent, under the old folder's lock: a connector an older build is still running there keeps it, and the start is refused as before. Adopting is idempotent; a profile that already has its own folder leaves the old one alone. Until a run adopts it, status, doctor, discard, release, import, shadow promote and guided setup's running check read the old folder. ResolveStateDir had no callers left since the worker's MCP server was removed (#815), and it accepted only the old name, so it goes. --- internal/commands/connect_doctor.go | 2 +- internal/commands/connect_operator.go | 17 +- internal/commands/connect_operator_test.go | 16 +- .../commands/connect_profile_state_test.go | 256 ++++++++++++++++++ internal/commands/connect_run.go | 51 +++- internal/commands/connect_setup_guided.go | 6 +- internal/commands/connect_start_wait_test.go | 4 +- internal/connector/ledger_dispatch.go | 95 ------- internal/connector/ledger_dispatch_test.go | 41 +-- internal/connector/lock.go | 11 +- internal/connector/lock_test.go | 2 - internal/connector/operator_migration_test.go | 4 +- internal/connector/statedir.go | 53 ++++ internal/connector/statedir_test.go | 141 ++++++++++ skills/basecamp-connect/SKILL.md | 4 +- 15 files changed, 517 insertions(+), 186 deletions(-) create mode 100644 internal/commands/connect_profile_state_test.go create mode 100644 internal/connector/statedir.go create mode 100644 internal/connector/statedir_test.go diff --git a/internal/commands/connect_doctor.go b/internal/commands/connect_doctor.go index 67609d173..29c70c4f2 100644 --- a/internal/commands/connect_doctor.go +++ b/internal/commands/connect_doctor.go @@ -124,7 +124,7 @@ func feedCheck(ctx context.Context, account *basecamp.AccountClient) setup.Check // gaps with their epoch ids, losses, the hold, and lifecycle messages that // wait for a decision. It reads the ledger read-only. func ledgerChecks(ctx context.Context, p connectProfile) []setup.Check { - dir, err := connectStatePath(p.file, false) + dir, err := connectStatePath(p.name, p.file, false) if err != nil { return []setup.Check{{Name: "Ledger", Status: setup.StatusFail, Message: errorMessage(err)}} } diff --git a/internal/commands/connect_operator.go b/internal/commands/connect_operator.go index ad4f71f19..d20748b9b 100644 --- a/internal/commands/connect_operator.go +++ b/internal/commands/connect_operator.go @@ -97,7 +97,7 @@ func parseEventIDArg(raw string) (int64, error) { // metadata beside it is written best-effort and says nothing on its own. func openConnectLedger(ctx context.Context, p connectProfile, requireStopped bool) (*connector.Ledger, func(), error) { done := func() {} - dir, err := connectStatePath(p.file, false) + dir, err := connectStatePath(p.name, p.file, false) if err != nil { return nil, done, err } @@ -205,7 +205,7 @@ func runConnectStatus(cmd *cobra.Command, shadow bool) error { if err != nil { return err } - dir, err := connectStatePath(p.file, shadow) + dir, err := connectStatePath(p.name, p.file, shadow) if err != nil { return err } @@ -513,7 +513,7 @@ to be discarded.`, if err != nil { return err } - shadowDir, err := connectStatePath(p.file, true) + shadowDir, err := connectStatePath(p.name, p.file, true) if err != nil { return err } @@ -521,13 +521,16 @@ to be discarded.`, if err != nil { return err } - parent, err := ensurePrivateChain(stateHome, "basecamp", "connect") - if err != nil { + if _, err := ensurePrivateChain(stateHome, "basecamp", "connect"); err != nil { return output.ErrUsage("The connector's state directory cannot be used: " + err.Error()) } + stateDir, err := connectStatePath(p.name, p.file, false) + if err != nil { + return err + } res, err := connector.PromoteShadow(cmd.Context(), connector.PromoteOptions{ ShadowDir: shadowDir, - StateDir: filepath.Join(parent, connector.StateDirName(p.file.AccountID, p.file.Agent.PersonID)), + StateDir: stateDir, AccountID: p.file.AccountID, AgentID: p.file.Agent.PersonID, By: operatorName(), @@ -583,7 +586,7 @@ The file is JSON: if err != nil { return output.ErrUsage(strings.TrimPrefix(err.Error(), "connector: ")) } - dir, err := connectStatePath(p.file, false) + dir, err := connectStatePath(p.name, p.file, false) if err != nil { return err } diff --git a/internal/commands/connect_operator_test.go b/internal/commands/connect_operator_test.go index 8719715c3..8db8a2eb6 100644 --- a/internal/commands/connect_operator_test.go +++ b/internal/commands/connect_operator_test.go @@ -57,7 +57,7 @@ func newOperatorFixture(t *testing.T) operatorFixture { // record 1 and a blocked record 2, and returns it open. func (f operatorFixture) ledger(t *testing.T, shadow bool) *connector.Ledger { t.Helper() - dir, err := connectStateDir(f.file, shadow) + dir, err := connectStateDir("agent", f.file, shadow) require.NoError(t, err) l, err := connector.OpenLedger(filepath.Join(dir, connector.LedgerFile)) require.NoError(t, err) @@ -155,7 +155,7 @@ func TestConnectDiscardAndRelease(t *testing.T) { require.NoError(t, err, out) assert.Contains(t, out, "Released") - dir, err := connectStatePath(f.file, false) + dir, err := connectStatePath("agent", f.file, false) require.NoError(t, err) l, err = connector.OpenLedgerReadOnly(context.Background(), filepath.Join(dir, connector.LedgerFile)) require.NoError(t, err) @@ -189,7 +189,7 @@ func TestConnectShadowPromoteAndImport(t *testing.T) { require.Error(t, err) assert.Equal(t, output.CodeUsage, usageError(t, err).Code) - dir, err := connectStatePath(f.file, false) + dir, err := connectStatePath("agent", f.file, false) require.NoError(t, err) lock, err := connector.AcquireInstanceLock(dir, f.file.AccountID, f.file.Agent.PersonID, time.Now()) require.NoError(t, err) @@ -301,7 +301,7 @@ func TestFakeMCPServer(t *testing.T) { func TestOperatorCommandsDoNotMigrateUnderARunningConnector(t *testing.T) { f := newOperatorFixture(t) require.NoError(t, f.ledger(t, false).Close()) - dir, err := connectStatePath(f.file, false) + dir, err := connectStatePath("agent", f.file, false) require.NoError(t, err) // A ledger an older binary wrote: its last migration is not recorded. @@ -325,7 +325,7 @@ func TestOperatorCommandsDoNotMigrateUnderARunningConnector(t *testing.T) { func TestTheMigrationGuardIsTheLockNotItsMetadata(t *testing.T) { f := newOperatorFixture(t) require.NoError(t, f.ledger(t, false).Close()) - dir, err := connectStatePath(f.file, false) + dir, err := connectStatePath("agent", f.file, false) require.NoError(t, err) db, err := sql.Open("sqlite", filepath.Join(dir, connector.LedgerFile)) require.NoError(t, err) @@ -348,7 +348,7 @@ func TestTheMigrationGuardIsTheLockNotItsMetadata(t *testing.T) { func TestImportRunsOnALedgerOlderThanTheBuild(t *testing.T) { f := newOperatorFixture(t) require.NoError(t, f.ledger(t, false).Close()) - dir, err := connectStatePath(f.file, false) + dir, err := connectStatePath("agent", f.file, false) require.NoError(t, err) db, err := sql.Open("sqlite", filepath.Join(dir, connector.LedgerFile)) require.NoError(t, err) @@ -365,7 +365,7 @@ func TestImportRunsOnALedgerOlderThanTheBuild(t *testing.T) { // itself over the lock it holds. _, err = f.run(t, output.FormatJSON, "import", file) if err != nil { - assert.NotContains(t, err.Error(), "already holds this account") + assert.NotContains(t, err.Error(), "already runs for this profile") assert.NotContains(t, err.Error(), "Stop the connector") return } @@ -396,7 +396,7 @@ func TestDoctorRefusesAShadowingToken(t *testing.T) { require.Error(t, err, args[0]) assert.Contains(t, err.Error(), "BASECAMP_TOKEN", args[0]) } - dir, err := connectStatePath(f.file, false) + dir, err := connectStatePath("agent", f.file, false) require.NoError(t, err) db, err := sql.Open("sqlite", filepath.Join(dir, connector.LedgerFile)) require.NoError(t, err) diff --git a/internal/commands/connect_profile_state_test.go b/internal/commands/connect_profile_state_test.go new file mode 100644 index 000000000..14d318c7a --- /dev/null +++ b/internal/commands/connect_profile_state_test.go @@ -0,0 +1,256 @@ +//go:build unix + +package commands + +import ( + "context" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/basecamp/basecamp-sdk/go/pkg/basecamp/eventfeed" + + "github.com/basecamp/basecamp-cli/internal/appctx" + "github.com/basecamp/basecamp-cli/internal/connector" + "github.com/basecamp/basecamp-cli/internal/connector/setup" + "github.com/basecamp/basecamp-cli/internal/output" +) + +type runningConnector struct { + out *lockedBuffer + done chan error + cancel context.CancelFunc +} + +func startProfileConnector(t *testing.T, s *connectSetupServer, profile string) runningConnector { + t.Helper() + app := newConnectSetupApp(t, s, profile) + ctx, cancel := context.WithCancel(t.Context()) + out := &lockedBuffer{} + app.Output = output.New(output.Options{Format: output.FormatJSON, Writer: out}) + cmd := NewConnectCmd() + cmd.SetArgs(nil) + cmd.SetContext(appctx.WithApp(ctx, app)) + cmd.SetOut(out) + cmd.SetErr(out) + cmd.SilenceErrors, cmd.SilenceUsage = true, true + r := runningConnector{out: out, done: make(chan error, 1), cancel: cancel} + go func() { r.done <- cmd.Execute() }() + t.Cleanup(func() { + cancel() + select { + case <-r.done: + case <-time.After(10 * time.Second): + t.Errorf("the connector for %s did not stop when canceled", profile) + } + }) + return r +} + +func (r runningConnector) awaitRunning(t *testing.T) { + t.Helper() + deadline := time.After(15 * time.Second) + for !strings.Contains(r.out.String(), "connector: running") { + select { + case err := <-r.done: + r.done <- err + t.Fatalf("the connector exited before it ran: %v\n%s", err, r.out.String()) + case <-deadline: + t.Fatalf("the connector never ran: %s", r.out.String()) + case <-time.After(20 * time.Millisecond): + } + } +} + +func (r runningConnector) awaitExit(t *testing.T) error { + t.Helper() + select { + case err := <-r.done: + r.done <- err + return err + case <-time.After(15 * time.Second): + t.Fatalf("the connector kept running: %s", r.out.String()) + return nil + } +} + +func setUpSecondProfile(t *testing.T, s *connectSetupServer, profile string) setup.File { + t.Helper() + out, err := runAgentConnect(t, newConnectSetupApp(t, s, profile)) + require.NoError(t, err, out) + out, err = runConnectSetupCmd(t, newConnectSetupApp(t, s, profile), "--operator", fmt.Sprint(setupOperatorPerson), serveArg()) + require.NoError(t, err, out) + file, err := setup.Load(connectSetupPath(t, profile)) + require.NoError(t, err) + return file +} + +func connectStateRoot(t *testing.T) string { + t.Helper() + root, err := connector.StateRoot() + require.NoError(t, err) + return root +} + +func connectRunKey(t *testing.T, s *connectSetupServer, file setup.File) eventfeed.CheckpointKey { + t.Helper() + origin, err := eventfeed.CanonicalOrigin(s.srv.URL) + require.NoError(t, err) + filters := eventfeed.Filters{ExcludePerformers: []int64{file.Agent.PersonID}, ActorTypes: []string{"person"}} + return eventfeed.CheckpointKey{Origin: origin, AccountID: file.AccountID, + ConsumerNamespace: connectConsumerNamespace(file.Agent.PersonID, false), FilterKey: filters.FilterKey()} +} + +const legacyPosition = "legacy-position-41" + +func seedLegacyLedger(t *testing.T, s *connectSetupServer, file setup.File) string { + t.Helper() + stateHome, err := connectStateHome() + require.NoError(t, err) + legacy, err := ensurePrivateChain(stateHome, "basecamp", "connect", connector.LegacyStateDirName(file.AccountID, file.Agent.PersonID)) + require.NoError(t, err) + l, err := connector.OpenLedger(filepath.Join(legacy, connector.LedgerFile)) + require.NoError(t, err) + ctx := context.Background() + _, err = l.RecordSeen(ctx, eventfeed.Event{ID: 41, Kind: "comment_created", EventType: "comment.created", Action: "created", + CreatedAt: time.Now(), BucketID: setupProject, CreatorID: setupOperatorPerson, RecordingID: 77}, connector.LanePoll) + require.NoError(t, err) + require.NoError(t, l.Save(ctx, connectRunKey(t, s, file), legacyPosition)) + require.NoError(t, l.Close()) + return legacy +} + +func assertLegacyLedgerKept(t *testing.T, dir string, s *connectSetupServer, file setup.File) { + t.Helper() + l, err := connector.OpenLedgerReadOnly(context.Background(), filepath.Join(dir, connector.LedgerFile)) + require.NoError(t, err) + defer func() { _ = l.Close() }() + position, ok, err := l.Load(context.Background(), connectRunKey(t, s, file)) + require.NoError(t, err) + require.True(t, ok, "the cursor came with the directory") + assert.Equal(t, legacyPosition, position) + _, found, err := l.Get(context.Background(), 41) + require.NoError(t, err) + assert.True(t, found, "the ledger's records came with it") +} + +func TestTwoProfilesForOneAgentEachRunAConnector(t *testing.T) { + f := newOperatorFixture(t) + other := setUpSecondProfile(t, f.s, "agent-b") + require.Equal(t, f.file.Agent.PersonID, other.Agent.PersonID, "both profiles are the same agent") + require.Equal(t, f.file.AccountID, other.AccountID) + + a := startProfileConnector(t, f.s, "agent") + a.awaitRunning(t) + b := startProfileConnector(t, f.s, "agent-b") + b.awaitRunning(t) + + root := connectStateRoot(t) + for _, profile := range []string{"agent", "agent-b"} { + dir := filepath.Join(root, connector.StateDirName(f.file.AccountID, f.file.Agent.PersonID, profile)) + _, err := os.Lstat(filepath.Join(dir, connector.LedgerFile)) + require.NoError(t, err, "%s runs on a ledger of its own", profile) + holder, ok := connector.InstanceHolder(dir, f.file.AccountID, f.file.Agent.PersonID) + require.True(t, ok, "%s holds the lock in its own state directory", profile) + assert.Equal(t, os.Getpid(), holder.PID) + } + + again := startProfileConnector(t, f.s, "agent") + err := again.awaitExit(t) + require.Error(t, err) + assert.Equal(t, output.CodeLockUnavailable, usageError(t, err).Code, again.out.String()) + assert.Contains(t, err.Error(), "already runs for this profile") +} + +func TestARunAdoptsTheLegacyStateDirectoryWithItsCursor(t *testing.T) { + f := newOperatorFixture(t) + root := connectStateRoot(t) + legacy := seedLegacyLedger(t, f.s, f.file) + + r := startProfileConnector(t, f.s, "agent") + r.awaitRunning(t) + require.Eventually(t, func() bool { return strings.Contains(r.out.String(), "resuming the feed from the stored position") }, + 10*time.Second, 20*time.Millisecond, "the run did not resume from the old cursor: %s", r.out.String()) + + dir := filepath.Join(root, connector.StateDirName(f.file.AccountID, f.file.Agent.PersonID, "agent")) + assert.Contains(t, r.out.String(), "state="+dir) + _, err := os.Lstat(legacy) + assert.ErrorIs(t, err, os.ErrNotExist, "the old directory moved rather than being copied") + r.cancel() + _ = r.awaitExit(t) + + assertLegacyLedgerKept(t, dir, f.s, f.file) +} + +func TestAStartRefusesWhileAnOlderConnectorHoldsTheLegacyDirectory(t *testing.T) { + f := newOperatorFixture(t) + root := connectStateRoot(t) + legacy := seedLegacyLedger(t, f.s, f.file) + lock, err := connector.AcquireInstanceLock(legacy, f.file.AccountID, f.file.Agent.PersonID, time.Now()) + require.NoError(t, err) + t.Cleanup(func() { _ = lock.Release() }) + + r := startProfileConnector(t, f.s, "agent") + err = r.awaitExit(t) + require.Error(t, err) + assert.Equal(t, output.CodeLockUnavailable, usageError(t, err).Code, r.out.String()) + _, err = os.Lstat(filepath.Join(legacy, connector.LedgerFile)) + require.NoError(t, err, "a running older connector keeps its directory") + _, err = os.Lstat(filepath.Join(root, connector.StateDirName(f.file.AccountID, f.file.Agent.PersonID, "agent"))) + assert.ErrorIs(t, err, os.ErrNotExist) +} + +func TestOperatorCommandsReadTheLegacyDirectoryUntilARunAdoptsIt(t *testing.T) { + f := newOperatorFixture(t) + root := connectStateRoot(t) + legacy := seedLegacyLedger(t, f.s, f.file) + + dir, err := connectStatePath("agent", f.file, false) + require.NoError(t, err) + assert.Equal(t, legacy, dir) + out, err := f.run(t, output.FormatJSON, "status") + require.NoError(t, err, out) + assert.Contains(t, out, `"profile": "agent"`) + _, err = os.Lstat(filepath.Join(root, connector.StateDirName(f.file.AccountID, f.file.Agent.PersonID, "agent"))) + assert.ErrorIs(t, err, os.ErrNotExist, "reading the state moves nothing") + + adopted, err := connectStateDir("agent", f.file, false) + require.NoError(t, err) + assert.Equal(t, filepath.Join(root, connector.StateDirName(f.file.AccountID, f.file.Agent.PersonID, "agent")), adopted) + dir, err = connectStatePath("agent", f.file, false) + require.NoError(t, err) + assert.Equal(t, adopted, dir) + + other, err := connectStatePath("agent-b", f.file, false) + require.NoError(t, err) + assert.Equal(t, filepath.Join(root, connector.StateDirName(f.file.AccountID, f.file.Agent.PersonID, "agent-b")), other, + "once adopted, the old directory is no other profile's") +} + +func TestAProfileThatCannotProveItIsTheAgentLeavesTheLegacyDirectory(t *testing.T) { + f := newOperatorFixture(t) + legacy := seedLegacyLedger(t, f.s, f.file) + app := newConnectSetupApp(t, f.s, "agent") + creds, err := app.Auth.GetStore().Load(app.Auth.CredentialKey()) + require.NoError(t, err) + creds.ExpiresAt = time.Now().Add(-time.Minute).Unix() + require.NoError(t, app.Auth.GetStore().Save(app.Auth.CredentialKey(), creds)) + f.s.mu.Lock() + f.s.refuseSecret = true + f.s.mu.Unlock() + + r := startProfileConnector(t, f.s, "agent") + err = r.awaitExit(t) + require.Error(t, err) + assert.Contains(t, err.Error(), "disconnected", r.out.String()) + _, err = os.Lstat(filepath.Join(legacy, connector.LedgerFile)) + require.NoError(t, err, "the old directory waits for the profile that is the agent") + _, err = os.Lstat(filepath.Join(connectStateRoot(t), connector.StateDirName(f.file.AccountID, f.file.Agent.PersonID, "agent"))) + assert.ErrorIs(t, err, os.ErrNotExist) +} diff --git a/internal/commands/connect_run.go b/internal/commands/connect_run.go index 345de3318..0cd9ff440 100644 --- a/internal/commands/connect_run.go +++ b/internal/commands/connect_run.go @@ -49,9 +49,6 @@ func addConnectRunFlags(cmd *cobra.Command, f *connectRunFlags) { fl.BoolVar(&f.hold, "hold", false, "Set the durable hold: intake and admission run, nothing is dispatched or posted until the hold is released, and earlier records wait for review") } -// connectStateHome is the directory holding the connector's state root, from -// connector.StateRoot so the connector and the worker's MCP server agree on -// one place. func connectStateHome() (string, error) { root, err := connector.StateRoot() if err != nil { @@ -78,37 +75,53 @@ func ensurePrivateChain(root string, parts ...string) (string, error) { return dir, nil } -// connectStateDir is the connector's state directory for a set-up profile, -// created owner-only: $XDG_STATE_HOME/basecamp/connect/-, or -// connect-shadow for a shadow run. Everything that reads the connector's -// state (status) resolves it here. -func connectStateDir(file setup.File, shadow bool) (string, error) { +func connectStateDir(profile string, file setup.File, shadow bool) (string, error) { stateHome, err := connectStateHome() if err != nil { return "", err } - return ensurePrivateChain(stateHome, connectStateParts(file, shadow)...) + parts := connectStateParts(profile, file, shadow) + parent, err := ensurePrivateChain(stateHome, parts[:len(parts)-1]...) + if err != nil { + return "", err + } + if _, err := connector.AdoptLegacyStateDir(parent, file.AccountID, file.Agent.PersonID, profile); err != nil { + return "", err + } + dir := filepath.Join(parent, parts[len(parts)-1]) + if err := setup.EnsurePrivateDir(dir); err != nil { + return "", err + } + return dir, nil } // connectStatePath is connectStateDir's path, created nothing: for commands // that only read the connector's state, and must not make a directory to do // it. -func connectStatePath(file setup.File, shadow bool) (string, error) { +func connectStatePath(profile string, file setup.File, shadow bool) (string, error) { stateHome, err := connectStateHome() if err != nil { return "", err } - return filepath.Join(append([]string{stateHome}, connectStateParts(file, shadow)...)...), nil + parts := connectStateParts(profile, file, shadow) + dir := filepath.Join(append([]string{stateHome}, parts...)...) + legacy := filepath.Join(filepath.Dir(dir), connector.LegacyStateDirName(file.AccountID, file.Agent.PersonID)) + if _, err := os.Lstat(dir); errors.Is(err, os.ErrNotExist) { + if _, err := os.Lstat(legacy); err == nil { + return legacy, nil + } + } + return dir, nil } -func connectStateParts(file setup.File, shadow bool) []string { +func connectStateParts(profile string, file setup.File, shadow bool) []string { group := "connect" if shadow { // An isolated ledger, lock and checkpoint: a shadow never shares a // position or a record with the connector it watches beside. group = "connect-shadow" } - return []string{"basecamp", group, connector.StateDirName(file.AccountID, file.Agent.PersonID)} + return []string{"basecamp", group, connector.StateDirName(file.AccountID, file.Agent.PersonID, profile)} } func runConnect(cmd *cobra.Command, f *connectRunFlags) error { @@ -175,7 +188,7 @@ func runConnect(cmd *cobra.Command, f *connectRunFlags) error { // until the credential is proven to be that agent, holding its lock // through the wait could keep the agent's real connector from starting. // The lock is taken for the run once the identity is verified. - stateDir, err := connectStateDir(file, f.shadow) + stateDir, err := connectStatePath(name, file, f.shadow) if err != nil { return output.ErrUsage("The connector's state directory cannot be used: " + err.Error()) } @@ -214,6 +227,13 @@ func runConnect(cmd *cobra.Command, f *connectRunFlags) error { return output.ErrAuth(err.Error()) } agentID := me.ID + stateDir, err = connectStateDir(name, file, f.shadow) + switch { + case errors.Is(err, connector.ErrAlreadyRunning): + return &output.Error{Code: output.CodeLockUnavailable, Message: err.Error()} + case err != nil: + return output.ErrUsage("The connector's state directory cannot be used: " + err.Error()) + } lock, err := acquireConnectLock(stateDir, account, agentID) if err != nil { return err @@ -413,6 +433,9 @@ func acquireConnectLock(stateDir, account string, agentID int64) (*connector.Ins // connectLockFree reports, by taking and at once releasing it, whether the // instance lock is free: the start's look before it waits for a token. func connectLockFree(stateDir, account string, agentID int64) error { + if _, err := os.Lstat(stateDir); errors.Is(err, os.ErrNotExist) { + return nil + } lock, err := acquireConnectLock(stateDir, account, agentID) if err != nil { return err diff --git a/internal/commands/connect_setup_guided.go b/internal/commands/connect_setup_guided.go index d3e68afb6..4fcd5b4b7 100644 --- a/internal/commands/connect_setup_guided.go +++ b/internal/commands/connect_setup_guided.go @@ -163,7 +163,7 @@ func runGuidedConnectSetup(cmd *cobra.Command, app *appctx.App, f *connectSetupF file = &loaded } - renderGuidedSummary(w, r, agent, *file, connectorRunning(*file), name) + renderGuidedSummary(w, r, agent, *file, connectorRunning(name, *file), name) return nil } @@ -543,8 +543,8 @@ func chooseGuidedProjects(w io.Writer, agent guidedAgent) ([]guidedProject, erro // connectorRunning reports whether a connector for this setup is running: // the instance lock's holder, as `connect status` reads it, is alive. -func connectorRunning(file setup.File) bool { - dir, err := connectStatePath(file, false) +func connectorRunning(profile string, file setup.File) bool { + dir, err := connectStatePath(profile, file, false) if err != nil { return false } diff --git a/internal/commands/connect_start_wait_test.go b/internal/commands/connect_start_wait_test.go index 901e4c2df..343594bbe 100644 --- a/internal/commands/connect_start_wait_test.go +++ b/internal/commands/connect_start_wait_test.go @@ -488,7 +488,7 @@ func TestAnOversizedServerFaultIsWaitedOut(t *testing.T) { // before the start waits for a token, not after. func TestASecondConnectorIsRefusedBeforeItWaitsForAToken(t *testing.T) { f := newOperatorFixture(t) - dir, err := connectStateDir(f.file, false) + dir, err := connectStateDir("agent", f.file, false) require.NoError(t, err) lock, err := connector.AcquireInstanceLock(dir, f.file.AccountID, f.file.Agent.PersonID, time.Now()) require.NoError(t, err) @@ -594,7 +594,7 @@ func TestAStartWaitingForItsTokenHoldsNoLock(t *testing.T) { require.Eventually(t, func() bool { return strings.Contains(out.String(), "could not get the agent's token yet") }, 10*time.Second, 20*time.Millisecond, "the start never waited: %s", out.String()) - dir, err := connectStateDir(f.file, false) + dir, err := connectStateDir("agent", f.file, false) require.NoError(t, err) lock, err := connector.AcquireInstanceLock(dir, f.file.AccountID, f.file.Agent.PersonID, time.Now()) require.NoError(t, err, "a waiting start held the agent's lock") diff --git a/internal/connector/ledger_dispatch.go b/internal/connector/ledger_dispatch.go index 18addbf81..50a3cc9c9 100644 --- a/internal/connector/ledger_dispatch.go +++ b/internal/connector/ledger_dispatch.go @@ -15,7 +15,6 @@ import ( "os" "path/filepath" "slices" - "strconv" "strings" "time" @@ -1287,56 +1286,6 @@ func scanAttributes(text string, pos int) (sgid string, end int, ok bool) { return sgid, pos, false } -// StateDirName is the connector's state directory for one account and agent, -// "-", inside StateRoot. -func StateDirName(accountID string, agentID int64) string { - return accountID + "-" + strconv.FormatInt(agentID, 10) -} - -// ErrNotAStateDir is a directory that is not a connector state directory for -// the account asked about. StateDirError carries why. -var ErrNotAStateDir = errors.New("not the connector's state directory for this account") - -// StateDirError says which rule a state directory failed, in fields a caller -// can build its own message from rather than by reading this one. -type StateDirError struct { - // Dir is the directory as given, made absolute. - Dir string - // Root is the connector's state root, the only place a state directory - // lives. - Root string - // Account is the account the directory names, empty when it names none; - // Want is the account it had to name. - Account, Want string - // Why is the rule it failed. - Why StateDirProblem -} - -// StateDirProblem is why a state directory was refused. -type StateDirProblem string - -const ( - // StateDirElsewhere is a directory outside the state root. - StateDirElsewhere StateDirProblem = "outside the connector's state root" - // StateDirMisnamed is a directory not named -. - StateDirMisnamed StateDirProblem = "not named -" - // StateDirOtherAccount is another account's state directory. - StateDirOtherAccount StateDirProblem = "another account's" -) - -func (e *StateDirError) Error() string { - switch e.Why { - case StateDirElsewhere: - return fmt.Sprintf("%s is not inside %s", e.Dir, e.Root) - case StateDirOtherAccount: - return fmt.Sprintf("%s belongs to account %s, not %s", e.Dir, e.Account, e.Want) - default: - return fmt.Sprintf("%s is not named -", e.Dir) - } -} - -func (e *StateDirError) Unwrap() error { return ErrNotAStateDir } - // StateRoot is where every connector state directory lives: // $XDG_STATE_HOME/basecamp/connect, or ~/.local/state/basecamp/connect when // XDG_STATE_HOME is unset or not absolute, as the XDG specification says. @@ -1352,49 +1301,5 @@ func StateRoot() (string, error) { return filepath.Join(filepath.Clean(base), "basecamp", "connect"), nil } -// ResolveStateDir is the one place a state directory is accepted: dir must -// be exactly StateRoot/-, and its account must be -// accountID, compared as numbers. It returns the directory made absolute — -// which is the one every caller should go on to use — and the agent's Person -// id. -// -// The location is part of the check, not only the name. A directory named -// for this account anywhere else — a copy of another account's ledger renamed -// to match — is refused, because the name is what binds a ledger to an -// account and anyone can choose a name. -func ResolveStateDir(dir, accountID string) (string, int64, error) { - root, err := StateRoot() - if err != nil { - return "", 0, err - } - abs, err := filepath.Abs(dir) - if err != nil { - return "", 0, fmt.Errorf("connector: state directory %q: %w", dir, err) - } - refuse := func(why StateDirProblem, account string) (string, int64, error) { - return "", 0, &StateDirError{Dir: abs, Root: root, Account: account, Want: accountID, Why: why} - } - if filepath.Dir(abs) != root { - return refuse(StateDirElsewhere, "") - } - account, agent, ok := strings.Cut(filepath.Base(abs), "-") - agentID, err := strconv.ParseInt(agent, 10, 64) - if !ok || err != nil || agentID <= 0 || agent != strconv.FormatInt(agentID, 10) { - // The agent is spelled one way, so one directory answers to one name: - // "+52007412" and "052007412" are other directories, not this one. - return refuse(StateDirMisnamed, account) - } - given, errGiven := strconv.ParseUint(account, 10, 64) - if errGiven != nil || given == 0 { - // Not an account at all: the name is wrong, not another account's. - return refuse(StateDirMisnamed, account) - } - want, errWant := strconv.ParseUint(accountID, 10, 64) - if errWant != nil || given != want { - return refuse(StateDirOtherAccount, account) - } - return abs, agentID, nil -} - // LedgerFile is the ledger's file name inside the state directory. const LedgerFile = "ledger.db" diff --git a/internal/connector/ledger_dispatch_test.go b/internal/connector/ledger_dispatch_test.go index 7042e2141..811f28591 100644 --- a/internal/connector/ledger_dispatch_test.go +++ b/internal/connector/ledger_dispatch_test.go @@ -840,51 +840,12 @@ func entityEncodedSGID(mention string) string { return mention[:i] + fmt.Sprintf("&#x%x;", mention[i]) + mention[i+1:] } -func TestResolveStateDirAcceptsOnlyTheCanonicalDirectory(t *testing.T) { +func TestStateRootFollowsXDGStateHome(t *testing.T) { home := t.TempDir() t.Setenv("XDG_STATE_HOME", home) root, err := StateRoot() require.NoError(t, err) assert.Equal(t, filepath.Join(home, "basecamp", "connect"), root) - canonical := filepath.Join(root, StateDirName("999", adapterAgentID)) - - resolved, agentID, err := ResolveStateDir(canonical, "999") - require.NoError(t, err) - assert.Equal(t, adapterAgentID, agentID) - assert.Equal(t, canonical, resolved) - resolved, agentID, err = ResolveStateDir(canonical+"/../"+StateDirName("999", adapterAgentID)+"/", "0999") - require.NoError(t, err, "accounts compare as numbers, and a trailing slash or a .. is the same directory") - assert.Equal(t, adapterAgentID, agentID) - assert.Equal(t, canonical, resolved, "the directory every caller goes on to use is the one that was checked") - - for name, dir := range map[string]string{ - "outside the root": filepath.Join(t.TempDir(), StateDirName("999", adapterAgentID)), - "nested in the root": filepath.Join(root, "x", StateDirName("999", adapterAgentID)), - "another account": filepath.Join(root, StateDirName("1000", adapterAgentID)), - "no agent": filepath.Join(root, "999-"), - "no account": filepath.Join(root, "-52007412"), - "not a number": filepath.Join(root, "999-abc"), - "a signed agent": filepath.Join(root, "999-+52007412"), - "a padded agent": filepath.Join(root, "999-052007412"), - "the root itself": root, - "above the root": filepath.Join(root, "..", StateDirName("999", adapterAgentID)), - } { - t.Run(name, func(t *testing.T) { - _, _, err := ResolveStateDir(dir, "999") - assert.ErrorIs(t, err, ErrNotAStateDir) - var refusal *StateDirError - require.ErrorAs(t, err, &refusal, "the refusal says why in fields, not in a message to be parsed") - assert.Equal(t, root, refusal.Root) - assert.Equal(t, "999", refusal.Want) - if name == "another account" { - assert.Equal(t, StateDirOtherAccount, refusal.Why) - assert.Equal(t, "1000", refusal.Account) - } - if name == "outside the root" { - assert.Equal(t, StateDirElsewhere, refusal.Why) - } - }) - } t.Setenv("XDG_STATE_HOME", "relative/state") root, err = StateRoot() diff --git a/internal/connector/lock.go b/internal/connector/lock.go index f153542b8..7dc298dca 100644 --- a/internal/connector/lock.go +++ b/internal/connector/lock.go @@ -12,17 +12,8 @@ import ( "github.com/basecamp/basecamp-cli/internal/connector/setup" ) -// ErrAlreadyRunning reports a second connector for the same agent. -var ErrAlreadyRunning = errors.New("connector: another connector already holds this account and agent") +var ErrAlreadyRunning = errors.New("connector: another connector already runs for this profile") -// InstanceLock is the refusal of a second connector on one agent identity. -// -// It is keyed on the ACCOUNT and the agent's Person id, not on the profile -// name. Two profiles can hold credentials for the same agent, and the failure -// this prevents is not "the same configuration twice" — it is one agent's -// mentions being dispatched twice, which is a property of the identity, not of -// the file that names it. -// // The kernel drops a flock when the holding descriptor closes, process death // included, so a crashed connector cannot wedge the lock and there is no // stale-lock reaping to get wrong. The metadata written beside it is diff --git a/internal/connector/lock_test.go b/internal/connector/lock_test.go index ff3852b93..736ac0f69 100644 --- a/internal/connector/lock_test.go +++ b/internal/connector/lock_test.go @@ -28,8 +28,6 @@ func TestSecondConnectorOnTheSameAgentIsRefused(t *testing.T) { assert.Contains(t, err.Error(), "pid", "the refusal says who is holding it") } -// The key is the identity, not the profile that names it: two profiles can -// hold credentials for one agent, and it is the agent that gets double-served. func TestTheLockIsKeyedOnAccountAndAgent(t *testing.T) { dir := t.TempDir() now := time.Now() diff --git a/internal/connector/operator_migration_test.go b/internal/connector/operator_migration_test.go index 2ccb12b22..850e6ddf1 100644 --- a/internal/connector/operator_migration_test.go +++ b/internal/connector/operator_migration_test.go @@ -34,8 +34,8 @@ const ( func shadowFixture(t *testing.T) (shadowDir, stateDir string) { t.Helper() root := filepath.Join(t.TempDir(), "basecamp") - shadowDir = filepath.Join(root, "connect-shadow", StateDirName(opAccount, opAgent)) - stateDir = filepath.Join(root, "connect", StateDirName(opAccount, opAgent)) + shadowDir = filepath.Join(root, "connect-shadow", StateDirName(opAccount, opAgent, "agent")) + stateDir = filepath.Join(root, "connect", StateDirName(opAccount, opAgent, "agent")) for _, dir := range []string{root, filepath.Dir(shadowDir), filepath.Dir(stateDir)} { require.NoError(t, os.MkdirAll(dir, 0o700)) require.NoError(t, os.Chmod(dir, 0o700)) diff --git a/internal/connector/statedir.go b/internal/connector/statedir.go new file mode 100644 index 000000000..4984176fb --- /dev/null +++ b/internal/connector/statedir.go @@ -0,0 +1,53 @@ +package connector + +import ( + "errors" + "fmt" + "os" + "path/filepath" + "strconv" + "time" +) + +func StateDirName(accountID string, agentID int64, profile string) string { + return LegacyStateDirName(accountID, agentID) + "-" + profile +} + +func LegacyStateDirName(accountID string, agentID int64) string { + return accountID + "-" + strconv.FormatInt(agentID, 10) +} + +func AdoptLegacyStateDir(parent, accountID string, agentID int64, profile string) (bool, error) { + dir := filepath.Join(parent, StateDirName(accountID, agentID, profile)) + legacy := filepath.Join(parent, LegacyStateDirName(accountID, agentID)) + if present, err := pathPresent(dir); err != nil || present { + return false, err + } + if present, err := pathPresent(legacy); err != nil || !present { + return false, err + } + lock, err := AcquireInstanceLock(legacy, accountID, agentID, time.Now()) + if err != nil { + return false, err + } + defer func() { _ = lock.Release() }() + if present, err := pathPresent(dir); err != nil || present { + return false, err + } + if err := os.Rename(legacy, dir); err != nil { + return false, fmt.Errorf("connector: move %s to %s: %w", legacy, dir, err) + } + _ = os.Remove(filepath.Join(dir, filepath.Base(lock.Path())+".json")) + return true, syncDirectory(parent) +} + +func pathPresent(path string) (bool, error) { + switch _, err := os.Lstat(path); { + case err == nil: + return true, nil + case errors.Is(err, os.ErrNotExist): + return false, nil + default: + return false, fmt.Errorf("connector: inspect %s: %w", path, err) + } +} diff --git a/internal/connector/statedir_test.go b/internal/connector/statedir_test.go new file mode 100644 index 000000000..3b965f8b0 --- /dev/null +++ b/internal/connector/statedir_test.go @@ -0,0 +1,141 @@ +package connector + +import ( + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func stateParent(t *testing.T) string { + t.Helper() + parent := filepath.Join(t.TempDir(), "connect") + require.NoError(t, os.Mkdir(parent, 0o700)) + return parent +} + +func legacyStateDir(t *testing.T, parent string) string { + t.Helper() + legacy := filepath.Join(parent, LegacyStateDirName("2914079", 52007412)) + require.NoError(t, os.Mkdir(legacy, 0o700)) + require.NoError(t, os.WriteFile(filepath.Join(legacy, LedgerFile), []byte("the old ledger"), 0o600)) + return legacy +} + +func TestStateDirNamesAreOnePerProfile(t *testing.T) { + assert.Equal(t, "2914079-52007412", LegacyStateDirName("2914079", 52007412)) + assert.Equal(t, "2914079-52007412-rob-agent", StateDirName("2914079", 52007412, "rob-agent")) + assert.NotEqual(t, StateDirName("2914079", 52007412, "a"), StateDirName("2914079", 52007412, "b")) + assert.NotEqual(t, StateDirName("2914079", 52007412, "1-a"), StateDirName("2914079", 520074121, "a")) +} + +func TestAdoptMovesTheLegacyDirectoryToTheProfile(t *testing.T) { + parent := stateParent(t) + legacy := legacyStateDir(t, parent) + + adopted, err := AdoptLegacyStateDir(parent, "2914079", 52007412, "agent-a") + require.NoError(t, err) + assert.True(t, adopted) + + dir := filepath.Join(parent, StateDirName("2914079", 52007412, "agent-a")) + data, err := os.ReadFile(filepath.Join(dir, LedgerFile)) + require.NoError(t, err) + assert.Equal(t, "the old ledger", string(data)) + _, err = os.Lstat(legacy) + assert.ErrorIs(t, err, os.ErrNotExist) + entries, err := os.ReadDir(dir) + require.NoError(t, err) + for _, e := range entries { + assert.NotEqual(t, ".json", filepath.Ext(e.Name()), "no holder is left behind for a lock nobody holds") + } + + again, err := AdoptLegacyStateDir(parent, "2914079", 52007412, "agent-a") + require.NoError(t, err) + assert.False(t, again, "adopting twice moves nothing") + data, err = os.ReadFile(filepath.Join(dir, LedgerFile)) + require.NoError(t, err) + assert.Equal(t, "the old ledger", string(data)) +} + +func TestOnlyOneProfileAdoptsTheLegacyDirectory(t *testing.T) { + parent := stateParent(t) + legacyStateDir(t, parent) + + first, err := AdoptLegacyStateDir(parent, "2914079", 52007412, "agent-a") + require.NoError(t, err) + require.True(t, first) + second, err := AdoptLegacyStateDir(parent, "2914079", 52007412, "agent-b") + require.NoError(t, err) + assert.False(t, second) + + _, err = os.Lstat(filepath.Join(parent, StateDirName("2914079", 52007412, "agent-b"))) + assert.ErrorIs(t, err, os.ErrNotExist, "the second profile starts a directory of its own") + _, err = os.Lstat(filepath.Join(parent, StateDirName("2914079", 52007412, "agent-a"), LedgerFile)) + require.NoError(t, err) +} + +func TestAdoptLeavesAProfileThatAlreadyHasADirectoryAlone(t *testing.T) { + parent := stateParent(t) + legacy := legacyStateDir(t, parent) + dir := filepath.Join(parent, StateDirName("2914079", 52007412, "agent-a")) + require.NoError(t, os.Mkdir(dir, 0o700)) + require.NoError(t, os.WriteFile(filepath.Join(dir, LedgerFile), []byte("the new ledger"), 0o600)) + + adopted, err := AdoptLegacyStateDir(parent, "2914079", 52007412, "agent-a") + require.NoError(t, err) + assert.False(t, adopted) + + data, err := os.ReadFile(filepath.Join(dir, LedgerFile)) + require.NoError(t, err) + assert.Equal(t, "the new ledger", string(data)) + data, err = os.ReadFile(filepath.Join(legacy, LedgerFile)) + require.NoError(t, err) + assert.Equal(t, "the old ledger", string(data)) +} + +func TestAdoptWithNoLegacyDirectoryCreatesNothing(t *testing.T) { + parent := stateParent(t) + + adopted, err := AdoptLegacyStateDir(parent, "2914079", 52007412, "agent-a") + require.NoError(t, err) + assert.False(t, adopted) + entries, err := os.ReadDir(parent) + require.NoError(t, err) + assert.Empty(t, entries) +} + +func TestAdoptRefusesWhileAConnectorRunsInTheLegacyDirectory(t *testing.T) { + parent := stateParent(t) + legacy := legacyStateDir(t, parent) + lock, err := AcquireInstanceLock(legacy, "2914079", 52007412, time.Now()) + require.NoError(t, err) + t.Cleanup(func() { _ = lock.Release() }) + + adopted, err := AdoptLegacyStateDir(parent, "2914079", 52007412, "agent-a") + require.ErrorIs(t, err, ErrAlreadyRunning) + assert.False(t, adopted) + _, err = os.Lstat(filepath.Join(legacy, LedgerFile)) + require.NoError(t, err) + _, err = os.Lstat(filepath.Join(parent, StateDirName("2914079", 52007412, "agent-a"))) + assert.ErrorIs(t, err, os.ErrNotExist) +} + +func TestTwoProfilesHoldTheirLocksAtOnceAndOneProfileOnlyOnce(t *testing.T) { + parent := stateParent(t) + now := time.Now() + a := filepath.Join(parent, StateDirName("2914079", 52007412, "agent-a")) + b := filepath.Join(parent, StateDirName("2914079", 52007412, "agent-b")) + + first, err := AcquireInstanceLock(a, "2914079", 52007412, now) + require.NoError(t, err) + t.Cleanup(func() { _ = first.Release() }) + other, err := AcquireInstanceLock(b, "2914079", 52007412, now) + require.NoError(t, err, "another profile for the same agent runs beside it") + t.Cleanup(func() { _ = other.Release() }) + + _, err = AcquireInstanceLock(a, "2914079", 52007412, now) + require.ErrorIs(t, err, ErrAlreadyRunning) +} diff --git a/skills/basecamp-connect/SKILL.md b/skills/basecamp-connect/SKILL.md index 6206e9f01..f34cfd3c5 100644 --- a/skills/basecamp-connect/SKILL.md +++ b/skills/basecamp-connect/SKILL.md @@ -58,7 +58,7 @@ Claude Code only. The connector runs on Linux and macOS. ### 1. Start it, and watch its output -Only one connector per agent can run at a time. If it's already running +Only one connector per profile can run at a time. If it's already running elsewhere, it refuses to start and says so: tell the person, and don't start a second. @@ -541,7 +541,7 @@ output and tell the person what it means: - **"was disconnected in Basecamp, or connected on another computer"**: run `basecamp connect setup -P ''` (it reconnects with one approval), then start it again. -- **Already running**: another connector for this agent holds its lock, on +- **Already running**: another connector for this profile holds its lock, on this computer or in another session. Don't start a second. - **`BASECAMP_TOKEN` is set**: the person unsets it, then you start it again. - **Anything else**: show the error, and run `basecamp connect doctor -P From 196057533deba2ed4d03506056e8d519d615455b Mon Sep 17 00:00:00 2001 From: Rob Zolkos Date: Tue, 13 Oct 2026 18:20:12 -0400 Subject: [PATCH 2/5] Refuse an operator decision whose state folder moved before its lock was taken A start can adopt the old - folder between an operator command resolving it and taking its lock. Taking the lock then recreates an empty folder, and import would apply to a new, empty ledger and report success. The ledger is looked for again under the lock. --- internal/commands/connect_operator.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/internal/commands/connect_operator.go b/internal/commands/connect_operator.go index d20748b9b..56ffa179c 100644 --- a/internal/commands/connect_operator.go +++ b/internal/commands/connect_operator.go @@ -132,6 +132,10 @@ func openConnectLedger(ctx context.Context, p connectProfile, requireStopped boo return nil, done, lockErr } done = func() { _ = lock.Release() } + if _, err := os.Lstat(path); err != nil { + done() + return nil, func() {}, output.ErrUsageHint(fmt.Sprintf("Profile %q's connector state moved while this command started", p.name), "Run the command again.") + } } ledger, err := connector.OpenLedger(path) //nolint:contextcheck // OpenLedger migrates on its own context if err != nil { From 5ad7d4fa6460d67b5cd7f8cfdcd4c6c8f8498acb Mon Sep 17 00:00:00 2001 From: Rob Zolkos Date: Tue, 13 Oct 2026 18:29:57 -0400 Subject: [PATCH 3/5] Operator writes adopt the old folder before they act Discard, release, import and shadow promote resolved the old - folder and acted on that path. A start adopting it in between left them holding a path that no longer named the ledger, and taking a lock there recreated an empty folder: promote could move the shadow ledger into it and orphan it. Each one now adopts the old folder for its profile first, under the old folder's lock, and acts on the profile's folder. While an older build's connector holds that lock, a decision acts on its ledger in place as before, and import and promote are refused as before. Reads (status, doctor, setup's running check) still read the old folder without moving it. --- internal/commands/connect_operator.go | 33 +++--- .../commands/connect_profile_state_test.go | 104 ++++++++++++++++++ internal/commands/connect_run.go | 23 +++- 3 files changed, 141 insertions(+), 19 deletions(-) diff --git a/internal/commands/connect_operator.go b/internal/commands/connect_operator.go index 56ffa179c..3004c4287 100644 --- a/internal/commands/connect_operator.go +++ b/internal/commands/connect_operator.go @@ -97,7 +97,10 @@ func parseEventIDArg(raw string) (int64, error) { // metadata beside it is written best-effort and says nothing on its own. func openConnectLedger(ctx context.Context, p connectProfile, requireStopped bool) (*connector.Ledger, func(), error) { done := func() {} - dir, err := connectStatePath(p.name, p.file, false) + dir, err := connectAdoptedStatePath(p.name, p.file, false) + if errors.Is(err, connector.ErrAlreadyRunning) { + dir, err = connectStatePath(p.name, p.file, false) + } if err != nil { return nil, done, err } @@ -517,10 +520,6 @@ to be discarded.`, if err != nil { return err } - shadowDir, err := connectStatePath(p.name, p.file, true) - if err != nil { - return err - } stateHome, err := connectStateHome() if err != nil { return err @@ -528,17 +527,21 @@ to be discarded.`, if _, err := ensurePrivateChain(stateHome, "basecamp", "connect"); err != nil { return output.ErrUsage("The connector's state directory cannot be used: " + err.Error()) } - stateDir, err := connectStatePath(p.name, p.file, false) - if err != nil { - return err + var res connector.PromoteResult + shadowDir, err := connectAdoptedStatePath(p.name, p.file, true) + var stateDir string + if err == nil { + stateDir, err = connectAdoptedStatePath(p.name, p.file, false) + } + if err == nil { + res, err = connector.PromoteShadow(cmd.Context(), connector.PromoteOptions{ + ShadowDir: shadowDir, + StateDir: stateDir, + AccountID: p.file.AccountID, + AgentID: p.file.Agent.PersonID, + By: operatorName(), + }) } - res, err := connector.PromoteShadow(cmd.Context(), connector.PromoteOptions{ - ShadowDir: shadowDir, - StateDir: stateDir, - AccountID: p.file.AccountID, - AgentID: p.file.Agent.PersonID, - By: operatorName(), - }) switch { case errors.Is(err, connector.ErrAlreadyRunning): return &output.Error{Code: output.CodeLockUnavailable, Message: err.Error(), diff --git a/internal/commands/connect_profile_state_test.go b/internal/commands/connect_profile_state_test.go index 14d318c7a..6d1cfdb05 100644 --- a/internal/commands/connect_profile_state_test.go +++ b/internal/commands/connect_profile_state_test.go @@ -254,3 +254,107 @@ func TestAProfileThatCannotProveItIsTheAgentLeavesTheLegacyDirectory(t *testing. _, err = os.Lstat(filepath.Join(connectStateRoot(t), connector.StateDirName(f.file.AccountID, f.file.Agent.PersonID, "agent"))) assert.ErrorIs(t, err, os.ErrNotExist) } + +func openLegacyLedger(t *testing.T, file setup.File, group string) (string, *connector.Ledger) { + t.Helper() + stateHome, err := connectStateHome() + require.NoError(t, err) + legacy, err := ensurePrivateChain(stateHome, "basecamp", group, connector.LegacyStateDirName(file.AccountID, file.Agent.PersonID)) + require.NoError(t, err) + l, err := connector.OpenLedger(filepath.Join(legacy, connector.LedgerFile)) + require.NoError(t, err) + return legacy, l +} + +func heldLegacyLedger(t *testing.T, file setup.File) string { + t.Helper() + legacy, l := openLegacyLedger(t, file, "connect") + _, err := l.SetHold(context.Background(), "local:tester", connector.HoldByOperator) + require.NoError(t, err) + require.NoError(t, l.Close()) + return legacy +} + +func assertNotHeld(t *testing.T, dir string) { + t.Helper() + l, err := connector.OpenLedgerReadOnly(context.Background(), filepath.Join(dir, connector.LedgerFile)) + require.NoError(t, err) + defer func() { _ = l.Close() }() + _, held, err := l.HoldMarker(context.Background()) + require.NoError(t, err) + assert.False(t, held, "the release reached this ledger") +} + +func TestADecisionAdoptsTheLegacyDirectoryBeforeItActs(t *testing.T) { + f := newOperatorFixture(t) + legacy := heldLegacyLedger(t, f.file) + + out, err := f.run(t, output.FormatJSON, "release") + require.NoError(t, err, out) + + dir := filepath.Join(connectStateRoot(t), connector.StateDirName(f.file.AccountID, f.file.Agent.PersonID, "agent")) + assertNotHeld(t, dir) + _, err = os.Lstat(legacy) + assert.ErrorIs(t, err, os.ErrNotExist) +} + +func TestADecisionBesideAnOlderConnectorActsOnItsLedger(t *testing.T) { + f := newOperatorFixture(t) + legacy := heldLegacyLedger(t, f.file) + lock, err := connector.AcquireInstanceLock(legacy, f.file.AccountID, f.file.Agent.PersonID, time.Now()) + require.NoError(t, err) + t.Cleanup(func() { _ = lock.Release() }) + + out, err := f.run(t, output.FormatJSON, "release") + require.NoError(t, err, out) + + assertNotHeld(t, legacy) + _, err = os.Lstat(filepath.Join(connectStateRoot(t), connector.StateDirName(f.file.AccountID, f.file.Agent.PersonID, "agent"))) + assert.ErrorIs(t, err, os.ErrNotExist) +} + +func TestAnImportBesideAnOlderConnectorIsRefused(t *testing.T) { + f := newOperatorFixture(t) + legacy := heldLegacyLedger(t, f.file) + lock, err := connector.AcquireInstanceLock(legacy, f.file.AccountID, f.file.Agent.PersonID, time.Now()) + require.NoError(t, err) + t.Cleanup(func() { _ = lock.Release() }) + file := filepath.Join(t.TempDir(), "reconciliation.json") + require.NoError(t, os.WriteFile(file, []byte(`{"version":1,"entries":[{"event_id":2,"decision":"done"}]}`), 0o600)) + + _, err = f.run(t, output.FormatJSON, "import", file) + require.Error(t, err) + assert.Equal(t, output.CodeLockUnavailable, usageError(t, err).Code) +} + +func TestPromoteAdoptsTheLegacyShadowAndPromotesIntoTheProfile(t *testing.T) { + f := newOperatorFixture(t) + legacyShadow, l := openLegacyLedger(t, f.file, "connect-shadow") + require.NoError(t, l.Close()) + + out, err := f.run(t, output.FormatJSON, "shadow", "promote") + require.NoError(t, err, out) + + root := connectStateRoot(t) + _, err = os.Lstat(filepath.Join(root, connector.StateDirName(f.file.AccountID, f.file.Agent.PersonID, "agent"), connector.LedgerFile)) + require.NoError(t, err, "the shadow ledger is the profile's connector ledger now") + _, err = os.Lstat(legacyShadow) + assert.ErrorIs(t, err, os.ErrNotExist) + _, err = os.Lstat(filepath.Join(root, connector.LegacyStateDirName(f.file.AccountID, f.file.Agent.PersonID))) + assert.ErrorIs(t, err, os.ErrNotExist, "nothing is promoted into the old layout") +} + +func TestPromoteBesideAnOlderShadowIsRefused(t *testing.T) { + f := newOperatorFixture(t) + legacyShadow, l := openLegacyLedger(t, f.file, "connect-shadow") + require.NoError(t, l.Close()) + lock, err := connector.AcquireInstanceLock(legacyShadow, f.file.AccountID, f.file.Agent.PersonID, time.Now()) + require.NoError(t, err) + t.Cleanup(func() { _ = lock.Release() }) + + _, err = f.run(t, output.FormatJSON, "shadow", "promote") + require.Error(t, err) + assert.Equal(t, output.CodeLockUnavailable, usageError(t, err).Code) + _, err = os.Lstat(filepath.Join(legacyShadow, connector.LedgerFile)) + require.NoError(t, err) +} diff --git a/internal/commands/connect_run.go b/internal/commands/connect_run.go index 0cd9ff440..22db3e06b 100644 --- a/internal/commands/connect_run.go +++ b/internal/commands/connect_run.go @@ -81,20 +81,35 @@ func connectStateDir(profile string, file setup.File, shadow bool) (string, erro return "", err } parts := connectStateParts(profile, file, shadow) - parent, err := ensurePrivateChain(stateHome, parts[:len(parts)-1]...) - if err != nil { + if _, err := ensurePrivateChain(stateHome, parts[:len(parts)-1]...); err != nil { return "", err } - if _, err := connector.AdoptLegacyStateDir(parent, file.AccountID, file.Agent.PersonID, profile); err != nil { + dir, err := connectAdoptedStatePath(profile, file, shadow) + if err != nil { return "", err } - dir := filepath.Join(parent, parts[len(parts)-1]) if err := setup.EnsurePrivateDir(dir); err != nil { return "", err } return dir, nil } +func connectAdoptedStatePath(profile string, file setup.File, shadow bool) (string, error) { + stateHome, err := connectStateHome() + if err != nil { + return "", err + } + dir := filepath.Join(append([]string{stateHome}, connectStateParts(profile, file, shadow)...)...) + parent := filepath.Dir(dir) + if _, err := os.Lstat(parent); errors.Is(err, os.ErrNotExist) { + return dir, nil + } + if _, err := connector.AdoptLegacyStateDir(parent, file.AccountID, file.Agent.PersonID, profile); err != nil { + return "", err + } + return dir, nil +} + // connectStatePath is connectStateDir's path, created nothing: for commands // that only read the connector's state, and must not make a directory to do // it. From f84eacba71478387ec4c6450004b46e74a9eea30 Mon Sep 17 00:00:00 2001 From: Rob Zolkos Date: Tue, 13 Oct 2026 18:42:55 -0400 Subject: [PATCH 4/5] Only a verified run moves the old folder; writers never recreate it Pi's second pass found the writer adoption in the previous commit let an operator command move the agent's history into whichever profile ran it, with no proof that profile is the agent, and that the fallback beside a held lock could still open a stale path and create a fresh ledger there. Back to one owner rule: only a run that has proved it is the agent moves the old folder. Decisions act on the folder status reads, old or new: one that takes no lock opens the ledger without creating it, so a path a start moved away from under it fails instead of making a new ledger, and one that takes the lock looks for the ledger again under it. Shadow promote moves the shadow ledger into the profile's own folder, never into the old one, and refuses while the old folder still holds the connector's ledger, as it refuses any existing ledger. --- internal/commands/connect_operator.go | 43 ++++++++++--------- .../commands/connect_profile_state_test.go | 27 +++++++++--- internal/commands/connect_run.go | 23 ++-------- 3 files changed, 47 insertions(+), 46 deletions(-) diff --git a/internal/commands/connect_operator.go b/internal/commands/connect_operator.go index 3004c4287..fd1e8e9d8 100644 --- a/internal/commands/connect_operator.go +++ b/internal/commands/connect_operator.go @@ -97,10 +97,7 @@ func parseEventIDArg(raw string) (int64, error) { // metadata beside it is written best-effort and says nothing on its own. func openConnectLedger(ctx context.Context, p connectProfile, requireStopped bool) (*connector.Ledger, func(), error) { done := func() {} - dir, err := connectAdoptedStatePath(p.name, p.file, false) - if errors.Is(err, connector.ErrAlreadyRunning) { - dir, err = connectStatePath(p.name, p.file, false) - } + dir, err := connectStatePath(p.name, p.file, false) if err != nil { return nil, done, err } @@ -140,7 +137,12 @@ func openConnectLedger(ctx context.Context, p connectProfile, requireStopped boo return nil, func() {}, output.ErrUsageHint(fmt.Sprintf("Profile %q's connector state moved while this command started", p.name), "Run the command again.") } } - ledger, err := connector.OpenLedger(path) //nolint:contextcheck // OpenLedger migrates on its own context + var ledger *connector.Ledger + if requireStopped || outOfDate { + ledger, err = connector.OpenLedger(path) //nolint:contextcheck // OpenLedger migrates on its own context + } else { + ledger, err = connector.OpenExistingLedger(ctx, path) + } if err != nil { done() return nil, func() {}, err @@ -524,24 +526,25 @@ to be discarded.`, if err != nil { return err } - if _, err := ensurePrivateChain(stateHome, "basecamp", "connect"); err != nil { - return output.ErrUsage("The connector's state directory cannot be used: " + err.Error()) + shadowDir, err := connectStatePath(p.name, p.file, true) + if err != nil { + return err } - var res connector.PromoteResult - shadowDir, err := connectAdoptedStatePath(p.name, p.file, true) - var stateDir string - if err == nil { - stateDir, err = connectAdoptedStatePath(p.name, p.file, false) + parent, err := ensurePrivateChain(stateHome, "basecamp", "connect") + if err != nil { + return output.ErrUsage("The connector's state directory cannot be used: " + err.Error()) } - if err == nil { - res, err = connector.PromoteShadow(cmd.Context(), connector.PromoteOptions{ - ShadowDir: shadowDir, - StateDir: stateDir, - AccountID: p.file.AccountID, - AgentID: p.file.Agent.PersonID, - By: operatorName(), - }) + legacy := filepath.Join(parent, connector.LegacyStateDirName(p.file.AccountID, p.file.Agent.PersonID), connector.LedgerFile) + if _, err := os.Lstat(legacy); err == nil { + return output.ErrUsage(fmt.Sprintf("%s: %s", legacy, connector.ErrLedgerExists)) } + res, err := connector.PromoteShadow(cmd.Context(), connector.PromoteOptions{ + ShadowDir: shadowDir, + StateDir: filepath.Join(parent, connector.StateDirName(p.file.AccountID, p.file.Agent.PersonID, p.name)), + AccountID: p.file.AccountID, + AgentID: p.file.Agent.PersonID, + By: operatorName(), + }) switch { case errors.Is(err, connector.ErrAlreadyRunning): return &output.Error{Code: output.CodeLockUnavailable, Message: err.Error(), diff --git a/internal/commands/connect_profile_state_test.go b/internal/commands/connect_profile_state_test.go index 6d1cfdb05..19d82786e 100644 --- a/internal/commands/connect_profile_state_test.go +++ b/internal/commands/connect_profile_state_test.go @@ -285,17 +285,16 @@ func assertNotHeld(t *testing.T, dir string) { assert.False(t, held, "the release reached this ledger") } -func TestADecisionAdoptsTheLegacyDirectoryBeforeItActs(t *testing.T) { +func TestADecisionActsOnTheLegacyLedgerWithoutMovingIt(t *testing.T) { f := newOperatorFixture(t) legacy := heldLegacyLedger(t, f.file) out, err := f.run(t, output.FormatJSON, "release") require.NoError(t, err, out) - dir := filepath.Join(connectStateRoot(t), connector.StateDirName(f.file.AccountID, f.file.Agent.PersonID, "agent")) - assertNotHeld(t, dir) - _, err = os.Lstat(legacy) - assert.ErrorIs(t, err, os.ErrNotExist) + assertNotHeld(t, legacy) + _, err = os.Lstat(filepath.Join(connectStateRoot(t), connector.StateDirName(f.file.AccountID, f.file.Agent.PersonID, "agent"))) + assert.ErrorIs(t, err, os.ErrNotExist, "only a run that proves it is the agent moves the old folder") } func TestADecisionBesideAnOlderConnectorActsOnItsLedger(t *testing.T) { @@ -327,7 +326,7 @@ func TestAnImportBesideAnOlderConnectorIsRefused(t *testing.T) { assert.Equal(t, output.CodeLockUnavailable, usageError(t, err).Code) } -func TestPromoteAdoptsTheLegacyShadowAndPromotesIntoTheProfile(t *testing.T) { +func TestPromoteMovesALegacyShadowLedgerIntoTheProfile(t *testing.T) { f := newOperatorFixture(t) legacyShadow, l := openLegacyLedger(t, f.file, "connect-shadow") require.NoError(t, l.Close()) @@ -338,12 +337,26 @@ func TestPromoteAdoptsTheLegacyShadowAndPromotesIntoTheProfile(t *testing.T) { root := connectStateRoot(t) _, err = os.Lstat(filepath.Join(root, connector.StateDirName(f.file.AccountID, f.file.Agent.PersonID, "agent"), connector.LedgerFile)) require.NoError(t, err, "the shadow ledger is the profile's connector ledger now") - _, err = os.Lstat(legacyShadow) + _, err = os.Lstat(filepath.Join(legacyShadow, connector.LedgerFile)) assert.ErrorIs(t, err, os.ErrNotExist) _, err = os.Lstat(filepath.Join(root, connector.LegacyStateDirName(f.file.AccountID, f.file.Agent.PersonID))) assert.ErrorIs(t, err, os.ErrNotExist, "nothing is promoted into the old layout") } +func TestPromoteRefusesWhileTheOldFolderHoldsTheConnectorsLedger(t *testing.T) { + f := newOperatorFixture(t) + legacy := seedLegacyLedger(t, f.s, f.file) + require.NoError(t, f.ledger(t, true).Close()) + + _, err := f.run(t, output.FormatJSON, "shadow", "promote") + require.Error(t, err) + assert.Equal(t, output.CodeUsage, usageError(t, err).Code) + assert.Contains(t, err.Error(), "already has a ledger") + assertLegacyLedgerKept(t, legacy, f.s, f.file) + _, err = os.Lstat(filepath.Join(connectStateRoot(t), connector.StateDirName(f.file.AccountID, f.file.Agent.PersonID, "agent"))) + assert.ErrorIs(t, err, os.ErrNotExist) +} + func TestPromoteBesideAnOlderShadowIsRefused(t *testing.T) { f := newOperatorFixture(t) legacyShadow, l := openLegacyLedger(t, f.file, "connect-shadow") diff --git a/internal/commands/connect_run.go b/internal/commands/connect_run.go index 22db3e06b..0cd9ff440 100644 --- a/internal/commands/connect_run.go +++ b/internal/commands/connect_run.go @@ -81,30 +81,15 @@ func connectStateDir(profile string, file setup.File, shadow bool) (string, erro return "", err } parts := connectStateParts(profile, file, shadow) - if _, err := ensurePrivateChain(stateHome, parts[:len(parts)-1]...); err != nil { - return "", err - } - dir, err := connectAdoptedStatePath(profile, file, shadow) + parent, err := ensurePrivateChain(stateHome, parts[:len(parts)-1]...) if err != nil { return "", err } - if err := setup.EnsurePrivateDir(dir); err != nil { - return "", err - } - return dir, nil -} - -func connectAdoptedStatePath(profile string, file setup.File, shadow bool) (string, error) { - stateHome, err := connectStateHome() - if err != nil { + if _, err := connector.AdoptLegacyStateDir(parent, file.AccountID, file.Agent.PersonID, profile); err != nil { return "", err } - dir := filepath.Join(append([]string{stateHome}, connectStateParts(profile, file, shadow)...)...) - parent := filepath.Dir(dir) - if _, err := os.Lstat(parent); errors.Is(err, os.ErrNotExist) { - return dir, nil - } - if _, err := connector.AdoptLegacyStateDir(parent, file.AccountID, file.Agent.PersonID, profile); err != nil { + dir := filepath.Join(parent, parts[len(parts)-1]) + if err := setup.EnsurePrivateDir(dir); err != nil { return "", err } return dir, nil From 8ead7320ab3efd16d3051cde492357f1009bab30 Mon Sep 17 00:00:00 2001 From: Rob Zolkos Date: Tue, 13 Oct 2026 18:52:01 -0400 Subject: [PATCH 5/5] Promote holds the old folder's lock and fails closed on it Pi's third pass: promote looked only for a ledger in the old - folder, so it went ahead beside an older connector that had taken its lock but not yet made its ledger, and read an old folder it could not inspect as one with no ledger. Promote now takes the old folder's lock for the whole promote when that folder exists, refuses while it is held, and refuses on anything but a clean "no ledger". --- internal/commands/connect_operator.go | 45 ++++++++++++++----- .../commands/connect_profile_state_test.go | 34 ++++++++++++++ 2 files changed, 69 insertions(+), 10 deletions(-) diff --git a/internal/commands/connect_operator.go b/internal/commands/connect_operator.go index fd1e8e9d8..42f0bc5f7 100644 --- a/internal/commands/connect_operator.go +++ b/internal/commands/connect_operator.go @@ -534,17 +534,18 @@ to be discarded.`, if err != nil { return output.ErrUsage("The connector's state directory cannot be used: " + err.Error()) } - legacy := filepath.Join(parent, connector.LegacyStateDirName(p.file.AccountID, p.file.Agent.PersonID), connector.LedgerFile) - if _, err := os.Lstat(legacy); err == nil { - return output.ErrUsage(fmt.Sprintf("%s: %s", legacy, connector.ErrLedgerExists)) + var res connector.PromoteResult + release, err := lockLegacyStateWithoutLedger(filepath.Join(parent, connector.LegacyStateDirName(p.file.AccountID, p.file.Agent.PersonID)), p.file) + if err == nil { + defer release() + res, err = connector.PromoteShadow(cmd.Context(), connector.PromoteOptions{ + ShadowDir: shadowDir, + StateDir: filepath.Join(parent, connector.StateDirName(p.file.AccountID, p.file.Agent.PersonID, p.name)), + AccountID: p.file.AccountID, + AgentID: p.file.Agent.PersonID, + By: operatorName(), + }) } - res, err := connector.PromoteShadow(cmd.Context(), connector.PromoteOptions{ - ShadowDir: shadowDir, - StateDir: filepath.Join(parent, connector.StateDirName(p.file.AccountID, p.file.Agent.PersonID, p.name)), - AccountID: p.file.AccountID, - AgentID: p.file.Agent.PersonID, - By: operatorName(), - }) switch { case errors.Is(err, connector.ErrAlreadyRunning): return &output.Error{Code: output.CodeLockUnavailable, Message: err.Error(), @@ -564,6 +565,30 @@ to be discarded.`, return cmd } +func lockLegacyStateWithoutLedger(legacy string, file setup.File) (func(), error) { + switch _, err := os.Lstat(legacy); { + case errors.Is(err, os.ErrNotExist): + return func() {}, nil + case err != nil: + return nil, err + } + lock, err := connector.AcquireInstanceLock(legacy, file.AccountID, file.Agent.PersonID, time.Now()) + if err != nil { + return nil, err + } + ledger := filepath.Join(legacy, connector.LedgerFile) + switch _, err := os.Lstat(ledger); { + case err == nil: + err = fmt.Errorf("connector: %s: %w", ledger, connector.ErrLedgerExists) + _ = lock.Release() + return nil, err + case !errors.Is(err, os.ErrNotExist): + _ = lock.Release() + return nil, err + } + return func() { _ = lock.Release() }, nil +} + // --- import --------------------------------------------------------------- // maxReconciliationBytes bounds a reconciliation file. diff --git a/internal/commands/connect_profile_state_test.go b/internal/commands/connect_profile_state_test.go index 19d82786e..1beeb7456 100644 --- a/internal/commands/connect_profile_state_test.go +++ b/internal/commands/connect_profile_state_test.go @@ -371,3 +371,37 @@ func TestPromoteBesideAnOlderShadowIsRefused(t *testing.T) { _, err = os.Lstat(filepath.Join(legacyShadow, connector.LedgerFile)) require.NoError(t, err) } + +func TestPromoteRefusesWhileAnOlderConnectorHoldsTheOldFolder(t *testing.T) { + f := newOperatorFixture(t) + require.NoError(t, f.ledger(t, true).Close()) + stateHome, err := connectStateHome() + require.NoError(t, err) + legacy, err := ensurePrivateChain(stateHome, "basecamp", "connect", connector.LegacyStateDirName(f.file.AccountID, f.file.Agent.PersonID)) + require.NoError(t, err) + lock, err := connector.AcquireInstanceLock(legacy, f.file.AccountID, f.file.Agent.PersonID, time.Now()) + require.NoError(t, err) + t.Cleanup(func() { _ = lock.Release() }) + + _, err = f.run(t, output.FormatJSON, "shadow", "promote") + require.Error(t, err) + assert.Equal(t, output.CodeLockUnavailable, usageError(t, err).Code) + _, err = os.Lstat(filepath.Join(connectStateRoot(t), connector.StateDirName(f.file.AccountID, f.file.Agent.PersonID, "agent"), connector.LedgerFile)) + assert.ErrorIs(t, err, os.ErrNotExist, "nothing was promoted beside a connector that is starting") +} + +func TestPromoteRefusesAnOldFolderItCannotInspect(t *testing.T) { + if os.Geteuid() == 0 { + t.Skip("root reads a directory whatever its mode") + } + f := newOperatorFixture(t) + require.NoError(t, f.ledger(t, true).Close()) + legacy := seedLegacyLedger(t, f.s, f.file) + require.NoError(t, os.Chmod(legacy, 0o000)) + t.Cleanup(func() { _ = os.Chmod(legacy, 0o700) }) + + _, err := f.run(t, output.FormatJSON, "shadow", "promote") + require.Error(t, err) + _, err = os.Lstat(filepath.Join(connectStateRoot(t), connector.StateDirName(f.file.AccountID, f.file.Agent.PersonID, "agent"), connector.LedgerFile)) + assert.ErrorIs(t, err, os.ErrNotExist, "an old ledger that cannot be seen is not taken for none") +}