Give a different account its own connection when reconnecting

docsgpt/connectors/service.py

@@ -809,7 +809,10 @@ def complete_oauth(conn, state_row: dict, provider: str, token_info: dict, accou
 
     Signing in to an account that already has a connection updates that
     connection (and drops the pending placeholder), so reconnecting from any
-    entry point heals every source and tool of the account.
+    entry point heals every source and tool of the account. Reconnecting a
+    connection but signing in to a different account never rewrites it (its
+    sources and tools would silently run as the new account): the new
+    account gets its own connection and the original keeps its status.
 
     Args:
         conn: Open connection inside a transaction.
@@ -817,17 +820,27 @@ def complete_oauth(conn, state_row: dict, provider: str, token_info: dict, accou
         provider: ``google_drive``, ``share_point`` or ``confluence``.
         token_info: Sanitised token info from the provider.
         account: The account's email or name, shown as "Connected as".
+
+    Returns:
+        The connection now holding the sign-in.
     """
     import uuid
 
     repo = ConnectorSessionsRepository(conn)
     target = state_row
+    definition = catalog.get_definition(provider)
+    display_name = definition.name if definition else provider
     existing = repo.find_account(state_row["user_id"], provider, server_url=None, account_label=account)
     if existing is not None and str(existing["id"]) != str(state_row["id"]):
         target = existing
         if not has_credentials(state_row) and normalize_status(state_row) == STATUS_PENDING:
             repo.delete_by_id(str(state_row["id"]))
-    definition = catalog.get_definition(provider)
+    elif existing is None and state_row.get("account_label"):
+        # The row belongs to another account (a reconnect that signed in as someone else).
+        target = repo.create(
+            state_row["user_id"], provider, connector_key=provider, auth_kind="oauth",
+            display_name=display_name, account_label=account, status=STATUS_PENDING,
+        ) or repo.find_account(state_row["user_id"], provider, server_url=None, account_label=account)
     write_secrets(
         conn,
         target,
@@ -839,7 +852,7 @@ def complete_oauth(conn, state_row: dict, provider: str, token_info: dict, accou
         account_label=account,
         connector_key=provider,
         auth_kind="oauth",
-        display_name=definition.name if definition else provider,
+        display_name=display_name,
         last_error=None,
     )
     resume_sources(conn, str(target["id"]))

tests/connectors/test_service_credentials.py

@@ -382,6 +382,37 @@ class TestOAuthCompletion:
         assert first["id"] != second["id"]
         assert len(service.list_connections(pg_conn, "alice")) == 2
 
+    def test_reconnect_as_the_same_account_heals_the_connection(self, pg_conn):
+        cid = _connection(pg_conn, status="reconnect_needed", secrets={"token_info": {"access_token": "old"}},
+                          account_label="a@example.com")
+        state_row = service.begin_oauth(pg_conn, "alice", "google_drive", cid)
+        row = service.complete_oauth(pg_conn, state_row, "google_drive", {"access_token": "new"}, "a@example.com")
+        assert str(row["id"]) == cid
+        assert row["status"] == "connected"
+
+    def test_reconnect_as_another_account_leaves_the_connection_alone(self, pg_conn):
+        cid = _connection(pg_conn, status="reconnect_needed", secrets={"token_info": {"access_token": "old"}},
+                          account_label="a@example.com")
+        source = _source(pg_conn, cid)
+        pg_conn.execute(
+            text("UPDATE sources SET metadata = '{\"sync_state\": \"paused_reconnect\"}' WHERE id = CAST(:i AS uuid)"),
+            {"i": source},
+        )
+        state_row = service.begin_oauth(pg_conn, "alice", "google_drive", cid)
+        row = service.complete_oauth(pg_conn, state_row, "google_drive", {"access_token": "b"}, "b@example.com")
+        assert str(row["id"]) != cid
+        assert row["account_label"] == "b@example.com"
+        assert row["status"] == "connected"
+        original = _row(pg_conn, cid)
+        assert original["account_label"] == "a@example.com"
+        assert original["user_email"] != "b@example.com"
+        assert original["status"] == "reconnect_needed"
+        assert service.read_secrets(original)["token_info"]["access_token"] == "old"
+        meta = pg_conn.execute(
+            text("SELECT metadata FROM sources WHERE id = CAST(:i AS uuid)"), {"i": source}
+        ).scalar()
+        assert meta["sync_state"] == "paused_reconnect"
+
     def test_reconnect_must_name_own_connection(self, pg_conn):
         cid = _connection(pg_conn, user="bob", secrets={})
         with pytest.raises(service.ConnectionUnavailable):